记忆体基类 (BaseMemory)¶

BaseMemory ¶

Bases: TFNeuralModel, ABC

Base class for memory.

记忆内目前主要有以下几种内容：

聊天记录
文档记录

针对以上两种内容，提供三种存储方式： 1. 纯文本存储（可以是基于内存的，也可以是基于PG） 2. 向量存储（向量模型可以自定义，存储数据库可以自定义） 3. 知识图谱存储（基于owl2的知识图谱存储，多种存储模式可选择）

记忆类的主要任务如下：

提供从记忆中获取内容的方法（依据自然语言输入，返回既定要求大小的内容） a. 对于交互而言，尽量使用query相关接口，基于message进行查询与召回 b. 与此同时，Memory也提供一系列Get方法，可以程序化的查询，方便实现一些管理接口，但与用户日常交互时尽量避免这种结构化接口的使用。
提供记忆的持久化与加载方法（这部分由基类TFBaseModel实现） a. 使用基于Message的Commit方法将信息持久化 b. 与此同时，也提供 add/update/delete 方法，可以允许通过结构化的数据调用来维护Memory内容，这些方法作为手动维护memory的补充，而不应该是主流方式。

recall `abstractmethod` ¶

recall(
    current_input: UserAndAssMsg,
    chunk_size: int | Annotated[list[int], Len(4, 4)],
    length_function: Callable[[str], int],
    exclude_str: Optional[str] = None,
) -> Tuple[
    Optional[list[UserAndAssMsg]],
    Optional[list[DocElement]],
    Optional[str],
]

Recalls the content from memory based on the given natural language input.

根据给定的自然语言输入从内存中检索内容。

Parameters:

Name	Type	Description	Default
`current_input`	`str`	The message user input. This is used to recall the content from memory. It is usually 用户输入的消息。用这个来从内存中回忆内容。通常由聊天输入历史记录格式化。	required
`chunk_size`	`Union[int, List[int]]`	The size of the content to recall. It can be a single number or a list of numbers. If it is a list, the first number is the size of the conversation content to recall, the second number is the size of the memo content to recall, the third number is the size of the knowledge content to recall, and the fourth number is the size of the keyword content to recall. If it is a single number, it will be converted into four equal numbers. 要回忆的内容的大小。可以是单个数字或数字列表。如果是列表，第一个数字是要回忆的对话内容的大小，第二个数字是要回忆的备忘内容的大小，第三个数字是要回忆的知识内容的大小，第四个数字是要回忆的关键字内容的大小。如果是单个数字，将被转换为四个相等的数字。	required
`length_function`	`Callable`	A function to calculate the length of the content to recall. 用于计算所回忆内容长度的函数。	required
`exclude_str`	`Optional[str]`	The string to exclude from the recall. 从回忆中排除的字符串。	`None`

Returns: Tuple[Optional[list[BaseMessage]], Optional[list[DocElement]], Optional[str]]: The content recalled from memory.

    分别是代表对话、文档元素和知识的内容。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def recall(
    self,
    current_input: UserAndAssMsg,
    chunk_size: int | Annotated[list[int], annotated_types.Len(4, 4)],  # type: ignore
    length_function: Callable[[str], int],
    exclude_str: Optional[str] = None,
) -> Tuple[Optional[list[UserAndAssMsg]], Optional[list[DocElement]], Optional[str]]:
    """
    Recalls the content from memory based on the given natural language input.

    根据给定的自然语言输入从内存中检索内容。

    Args:
        current_input (str): The message user input. This is used to recall the content from memory. It is usually

            用户输入的消息。用这个来从内存中回忆内容。通常由聊天输入历史记录格式化。

        chunk_size (Union[int, List[int]]): The size of the content to recall. It can be a single number or a list
            of numbers. If it is a list, the first number is the size of the conversation content to recall, the
            second number is the size of the memo content to recall, the third number is the size of the knowledge
            content to recall, and the fourth number is the size of the keyword content to recall. If it is a single
            number, it will be converted into four equal numbers.

            要回忆的内容的大小。可以是单个数字或数字列表。如果是列表，第一个数字是要回忆的对话内容的大小，第二个数字是要回忆的备忘内容的大小，
            第三个数字是要回忆的知识内容的大小，第四个数字是要回忆的关键字内容的大小。如果是单个数字，将被转换为四个相等的数字。

        length_function (Callable): A function to calculate the length of the content to recall.

            用于计算所回忆内容长度的函数。

        exclude_str (Optional[str]): The string to exclude from the recall.

            从回忆中排除的字符串。
    Returns:
        Tuple[Optional[list[BaseMessage]], Optional[list[DocElement]], Optional[str]]: The content recalled from
            memory.

            分别是代表对话、文档元素和知识的内容。
    """
    ...

async_recall `abstractmethod` `async` ¶

async_recall(
    current_input: UserAndAssMsg,
    chunk_size: int | Annotated[list[int], Len(4, 4)],
    length_function: Callable[[str], int],
    exclude_str: Optional[str] = None,
) -> Tuple[
    Optional[list[UserAndAssMsg]],
    Optional[list[DocElement]],
    Optional[str],
]

Async Recalls the content from memory based on the given natural language input.

根据给定的自然语言输入从内存中使用Asycn模式检索内容。

Parameters:

Name	Type	Description	Default
`current_input`	`str`	The message user input. This is used to recall the content from memory. It is usually 用户输入的消息。用这个来从内存中回忆内容。通常由聊天输入历史记录格式化。	required
`chunk_size`	`Union[int, List[int]]`	The size of the content to recall. It can be a single number or a list of numbers. If it is a list, the first number is the size of the conversation content to recall, the second number is the size of the memo content to recall, and the third number is the size of the knowledge content to recall. If it is a single number, it will be converted into three equal numbers. 要回忆的内容的大小。可以是单个数字或数字列表。如果是列表，第一个数字是要回忆的对话内容的大小，第二个数字是要回忆的备忘内容的大小，第三个数字是要回忆的知识内容的大小。如果是单个数字，将被转换为三个相等的数字。	required
`length_function`	`Callable`	A function to calculate the length of the content to recall. 用于计算所回忆内容长度的函数。	required
`exclude_str`	`Optional[str]`	The string to exclude from the recall. 从回忆中排除的字符串。	`None`

Returns: Tuple[Optional[list[BaseMessage]], Optional[list[DocElement]], Optional[str]]: The content recalled from memory.

    分别是代表对话、文档元素和知识的内容。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def async_recall(
    self,
    current_input: UserAndAssMsg,
    chunk_size: int | Annotated[list[int], annotated_types.Len(4, 4)],  # type: ignore
    length_function: Callable[[str], int],
    exclude_str: Optional[str] = None,
) -> Tuple[Optional[list[UserAndAssMsg]], Optional[list[DocElement]], Optional[str]]:
    """
    Async Recalls the content from memory based on the given natural language input.

    根据给定的自然语言输入从内存中使用Asycn模式检索内容。

    Args:
        current_input (str): The message user input. This is used to recall the content from memory. It is usually

            用户输入的消息。用这个来从内存中回忆内容。通常由聊天输入历史记录格式化。

        chunk_size (Union[int, List[int]]): The size of the content to recall. It can be a single number or a list
            of numbers. If it is a list, the first number is the size of the conversation content to recall, the
            second number is the size of the memo content to recall, and the third number is the size of the
            knowledge content to recall. If it is a single number, it will be converted into three equal numbers.

            要回忆的内容的大小。可以是单个数字或数字列表。如果是列表，第一个数字是要回忆的
            对话内容的大小，第二个数字是要回忆的备忘内容的大小，第三个数字是要回忆的知识内容的大小。如果是单个数字，将被转换为三个相等的数字。

        length_function (Callable): A function to calculate the length of the content to recall.

            用于计算所回忆内容长度的函数。

        exclude_str (Optional[str]): The string to exclude from the recall.

            从回忆中排除的字符串。
    Returns:
        Tuple[Optional[list[BaseMessage]], Optional[list[DocElement]], Optional[str]]: The content recalled from
            memory.

            分别是代表对话、文档元素和知识的内容。
    """
    ...

commit `abstractmethod` ¶

commit(msg: UserAndAssMsg) -> None

Commits the chain result to memory.

将链结果提交到记忆中。

一般可以在此进行比如知识更新，或者文档更新相关操作。

Parameters:

Name	Type	Description	Default
`msg`	`UserAndAssMsg`	The chain result to commit. \| 要提交的链结果。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def commit(self, msg: UserAndAssMsg) -> None:
    """
    Commits the chain result to memory.

    将链结果提交到记忆中。

    一般可以在此进行比如知识更新，或者文档更新相关操作。

    Args:
        msg (UserAndAssMsg): The chain result to commit. | 要提交的链结果。
    """
    ...

acommit `abstractmethod` `async` ¶

acommit(msg: UserAndAssMsg) -> None

Commits the chain result to memory asynchronously.

Default call self.commit()

Parameters:

Name	Type	Description	Default
`msg`	`UserAndAssMsg`	The chain result to commit. \| 要提交的链结果。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def acommit(self, msg: UserAndAssMsg) -> None:
    """
    Commits the chain result to memory asynchronously.

    Default call self.commit()

    Args:
        msg (UserAndAssMsg): The chain result to commit. | 要提交的链结果。
    """
    ...

sync_msg ¶

sync_msg(msg: UserAndAssMsg) -> None

将 msg 上的运行时元数据（intermediate_trace、token_usage 等）同步回存储。

与 commit 的区别：commit 是首次写入完整消息；sync_msg 是对已 commit 消息的运行时字段回写。与 edit_msg 的区别：edit_msg 由用户触发，更新 content/text 并联动索引；sync_msg 由系统触发，仅更新非内容字段，不触发索引。

默认 no-op，子类按需覆盖。

Source code in tfrobot/brain/memory/base.py

def sync_msg(self, msg: UserAndAssMsg) -> None:
    """
    将 msg 上的运行时元数据（intermediate_trace、token_usage 等）同步回存储。

    与 commit 的区别：commit 是首次写入完整消息；sync_msg 是对已 commit 消息的运行时字段回写。
    与 edit_msg 的区别：edit_msg 由用户触发，更新 content/text 并联动索引；sync_msg 由系统触发，仅更新非内容字段，不触发索引。

    默认 no-op，子类按需覆盖。
    """
    pass

async_sync_msg `async` ¶

async_sync_msg(msg: UserAndAssMsg) -> None

sync_msg 的异步版本。默认委托同步实现。

Source code in tfrobot/brain/memory/base.py

async def async_sync_msg(self, msg: UserAndAssMsg) -> None:
    """sync_msg 的异步版本。默认委托同步实现。"""
    self.sync_msg(msg)

get_platforms `abstractmethod` ¶

get_platforms() -> Sequence[tuple[PLATFORM_ID, str]] | None

获取当前会话平台列表，返回一个 list[tuple] tuple第一个元素为ID，第二个元素为platform可读标识

但注意，有的ConversationManager，如 DictBaseConversationManger 无Platform概念，因此可能返回None

Notes

返回值如果为None，表示当前Memory不支持Platform。返回值如果为List，表示当前Memory支持Platform模式，即使为空，也支持创建。这种表征方法并不是非常明确，更好的做法是添加一个 is_support_platform bool字段，但未来开发新的Memory建议都需要实现platform管理返回None仅仅是为了向前兼容 AIMemory+DictBaseConversationManager这种极简情况，考虑到Memory中的接口寸土寸金，因此不添加专门用于表征是否支持Platform的字段

Returns:

Type	Description
`Sequence[tuple[PLATFORM_ID, str]] \| None`	list[tuple[int, str]] \| None: 注意返回值可空

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_platforms(self) -> Sequence[tuple[PLATFORM_ID, str]] | None:
    """
    获取当前会话平台列表，返回一个 list[tuple] tuple第一个元素为ID，第二个元素为platform可读标识

    但注意，有的ConversationManager，如 DictBaseConversationManger 无Platform概念，因此可能返回None

    Notes:
        返回值如果为None，表示当前Memory不支持Platform。
        返回值如果为List，表示当前Memory支持Platform模式，即使为空，也支持创建。
        这种表征方法并不是非常明确，更好的做法是添加一个 is_support_platform bool字段，但未来开发新的Memory建议都需要实现platform管理
        返回None仅仅是为了向前兼容 AIMemory+DictBaseConversationManager这种极简情况，考虑到Memory中的接口寸土寸金，因此不添加专门用于表征是否支持Platform的字段

    Returns:
        list[tuple[int, str]] | None: 注意返回值可空
    """
    ...

aget_platforms `abstractmethod` `async` ¶

aget_platforms() -> (
    Sequence[tuple[PLATFORM_ID, str]] | None
)

获取当前会话平台列表，返回一个 list[tuple] tuple第一个元素为ID，第二个元素为platform可读标识

但注意，有的ConversationManager，如 DictBaseConversationManger 无Platform概念，因此可能返回None

Notes

返回值如果为None，表示当前Memory不支持Platform。返回值如果为List，表示当前Memory支持Platform模式，即使为空，也支持创建。这种表征方法并不是非常明确，更好的做法是添加一个 is_support_platform bool字段，但未来开发新的Memory建议都需要实现platform管理返回None仅仅是为了向前兼容 AIMemory+DictBaseConversationManager这种极简情况，考虑到Memory中的接口寸土寸金，因此不添加专门用于表征是否支持Platform的字段

Returns:

Type	Description
`Sequence[tuple[PLATFORM_ID, str]] \| None`	list[tuple[int, str]] \| None: 注意返回值可空

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_platforms(self) -> Sequence[tuple[PLATFORM_ID, str]] | None:
    """
    获取当前会话平台列表，返回一个 list[tuple] tuple第一个元素为ID，第二个元素为platform可读标识

    但注意，有的ConversationManager，如 DictBaseConversationManger 无Platform概念，因此可能返回None

    Notes:
        返回值如果为None，表示当前Memory不支持Platform。
        返回值如果为List，表示当前Memory支持Platform模式，即使为空，也支持创建。
        这种表征方法并不是非常明确，更好的做法是添加一个 is_support_platform bool字段，但未来开发新的Memory建议都需要实现platform管理
        返回None仅仅是为了向前兼容 AIMemory+DictBaseConversationManager这种极简情况，考虑到Memory中的接口寸土寸金，因此不添加专门用于表征是否支持Platform的字段

    Returns:
        list[tuple[int, str]] | None: 注意返回值可空
    """
    ...

add_platform `abstractmethod` ¶

add_platform(platform_name: str) -> PLATFORM_ID

添加Platform，返回值为PlatformID

Parameters:

Name	Type	Description	Default
`platform_name`	`str`	Platform名称，但需要注意，有些Memory可能会对PlatformName做Reformat	required

Returns:

Name	Type	Description
`PLATFORM_ID`	`PLATFORM_ID`	PlatformID

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_platform(self, platform_name: str) -> PLATFORM_ID:
    """
    添加Platform，返回值为PlatformID

    Args:
        platform_name (str): Platform名称，但需要注意，有些Memory可能会对PlatformName做Reformat

    Returns:
        PLATFORM_ID: PlatformID
    """
    ...

aadd_platform `abstractmethod` `async` ¶

aadd_platform(platform_name: str) -> PLATFORM_ID

添加Platform，返回值为PlatformID

Parameters:

Name	Type	Description	Default
`platform_name`	`str`	Platform名称，但需要注意，有些Memory可能会对PlatformName做Reformat	required

Returns:

Name	Type	Description
`PLATFORM_ID`	`PLATFORM_ID`	PlatformID

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_platform(self, platform_name: str) -> PLATFORM_ID:
    """
    添加Platform，返回值为PlatformID

    Args:
        platform_name (str): Platform名称，但需要注意，有些Memory可能会对PlatformName做Reformat

    Returns:
        PLATFORM_ID: PlatformID
    """
    ...

update_platform `abstractmethod` ¶

update_platform(
    platform_id: PLATFORM_ID, platform_name: str
) -> None

更新Platform名称 | Update platform name

Parameters:

Name	Type	Description	Default
`platform_id`	`PLATFORM_ID`	Platform的ID \| Platform ID	required
`platform_name`	`str`	新的Platform名称 \| New platform name	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_platform(self, platform_id: PLATFORM_ID, platform_name: str) -> None:
    """
    更新Platform名称 | Update platform name

    Args:
        platform_id (PLATFORM_ID): Platform的ID | Platform ID
        platform_name (str): 新的Platform名称 | New platform name
    """
    ...

aupdate_platform `abstractmethod` `async` ¶

aupdate_platform(
    platform_id: PLATFORM_ID, platform_name: str
) -> None

更新Platform名称 | Update platform name

Parameters:

Name	Type	Description	Default
`platform_id`	`PLATFORM_ID`	Platform的ID \| Platform ID	required
`platform_name`	`str`	新的Platform名称 \| New platform name	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_platform(self, platform_id: PLATFORM_ID, platform_name: str) -> None:
    """
    更新Platform名称 | Update platform name

    Args:
        platform_id (PLATFORM_ID): Platform的ID | Platform ID
        platform_name (str): 新的Platform名称 | New platform name
    """
    ...

delete_platform `abstractmethod` ¶

delete_platform(platform_id: PLATFORM_ID) -> None

删除Platform | Delete platform

Parameters:

Name	Type	Description	Default
`platform_id`	`PLATFORM_ID`	要删除的Platform的ID \| Platform ID to delete	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_platform(self, platform_id: PLATFORM_ID) -> None:
    """
    删除Platform | Delete platform

    Args:
        platform_id (PLATFORM_ID): 要删除的Platform的ID | Platform ID to delete
    """
    ...

adelete_platform `abstractmethod` `async` ¶

adelete_platform(platform_id: PLATFORM_ID) -> None

删除Platform | Delete platform

Parameters:

Name	Type	Description	Default
`platform_id`	`PLATFORM_ID`	要删除的Platform的ID \| Platform ID to delete	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_platform(self, platform_id: PLATFORM_ID) -> None:
    """
    删除Platform | Delete platform

    Args:
        platform_id (PLATFORM_ID): 要删除的Platform的ID | Platform ID to delete
    """
    ...

get_conversations `abstractmethod` ¶

get_conversations(
    cursor: Optional[str] = None,
    count: Optional[int] = None,
    platform_id: Optional[PLATFORM_ID] = None,
) -> tuple[list[tuple[CONVERSATION_KEY, str]], str]

Get all conversations from memory.

从记忆中获取所有对话。

因为会话排序随时有可能打乱，因此每次请求的时候指定一个初始位置是很有必要的。如此一来，前台可以获取到最新并且没有重复的对话。这里的最佳实践本来应该使用游标，比如使用 update_timestamp * 1000 + id % 1000 生成游标。但我们系统并非专业的会话管理系统，其数据压力和数据量都不会过大，因此这里不单独维护游标字段，而是直接让前台动态使用当前最后一个对话的ID作为游标。这样可以保证前台获取到的对话是最新的，且不会重复。如果有问题，刷新页面即可解决。

count支持负值，表示从后向前取数。

Parameters:

Name	Type	Description	Default
`cursor`	`int \| str \| None`	The id/index of the conversation to start from. \| 要从哪个对话开始。	`None`
`count`	`Optional[int]`	The number of conversations to get. \| 要获取的对话数量。	`None`
`platform_id`	`Optional[int]`	The platform id of the conversation. \| 对话的平台ID。	`None`

Returns:

Type	Description
`tuple[list[tuple[CONVERSATION_KEY, str]], str]`	tuple[list[tuple[CONVERSATION_KEY, str]], str]: The list of conversations and the next cursor. \| 对话列表和下一个游标。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_conversations(
    self, cursor: Optional[str] = None, count: Optional[int] = None, platform_id: Optional[PLATFORM_ID] = None
) -> tuple[list[tuple[CONVERSATION_KEY, str]], str]:
    """
    Get all conversations from memory.

    从记忆中获取所有对话。

    因为会话排序随时有可能打乱，因此每次请求的时候指定一个初始位置是很有必要的。如此一来，前台可以获取到最新并且没有重复的对话。这里的最佳实践本来应该
    使用游标，比如使用 update_timestamp * 1000 + id % 1000 生成游标。但我们系统并非专业的会话管理系统，其数据压力和数据量都不会过大，因此这里
    不单独维护游标字段，而是直接让前台动态使用当前最后一个对话的ID作为游标。这样可以保证前台获取到的对话是最新的，且不会重复。如果有问题，刷新页面即可解决。

    count支持负值，表示从后向前取数。

    Args:
        cursor (int | str | None): The id/index of the conversation to start from. | 要从哪个对话开始。
        count (Optional[int]): The number of conversations to get. | 要获取的对话数量。
        platform_id (Optional[int]): The platform id of the conversation. | 对话的平台ID。

    Returns:
        tuple[list[tuple[CONVERSATION_KEY, str]], str]: The list of conversations and the next cursor. | 对话列表和下一个游标。
    """
    ...

aget_conversations `abstractmethod` `async` ¶

aget_conversations(
    cursor: Optional[str] = None,
    count: Optional[int] = None,
    platform_id: Optional[PLATFORM_ID] = None,
) -> tuple[list[tuple[CONVERSATION_KEY, str]], str]

Get all conversations from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_conversations(
    self, cursor: Optional[str] = None, count: Optional[int] = None, platform_id: Optional[PLATFORM_ID] = None
) -> tuple[list[tuple[CONVERSATION_KEY, str]], str]:
    """Get all conversations from memory asynchronously."""
    ...

get_conversation `abstractmethod` ¶

get_conversation(
    conversation_id: CONVERSATION_KEY,
) -> Optional[BaseBufferStore]

Get a conversation from memory.

从记忆中获取一段对话。

注意页码可以是负数，表示从后向前取第几页。

Parameters:

Name	Type	Description	Default
`conversation_id`	`int \| str`	The id/index of the conversation. \| 对话的索引。	required

Returns:

Type	Description
`Optional[BaseBufferStore]`	list[BaseMessage]: The conversation. \| 对话。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_conversation(self, conversation_id: CONVERSATION_KEY) -> Optional[BaseBufferStore]:
    """
    Get a conversation from memory.

    从记忆中获取一段对话。

    注意页码可以是负数，表示从后向前取第几页。

    Args:
        conversation_id (int | str): The id/index of the conversation. | 对话的索引。

    Returns:
        list[BaseMessage]: The conversation. | 对话。
    """
    ...

aget_conversation `abstractmethod` `async` ¶

aget_conversation(
    conversation_id: CONVERSATION_KEY,
) -> Optional[BaseBufferStore]

Get a conversation from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_conversation(self, conversation_id: CONVERSATION_KEY) -> Optional[BaseBufferStore]:
    """Get a conversation from memory asynchronously."""
    ...

get_conversation_messages `abstractmethod` ¶

get_conversation_messages(
    conversation_id: CONVERSATION_KEY,
    page: int,
    size: int,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[
        Callable[[Attributes], bool]
    ] = None,
) -> list[UserAndAssMsg]

Get messages from a conversation.

Parameters:

Name	Type	Description	Default
`conversation_id`	`int \| str`	The id/index of the conversation. \| 对话的索引。	required
`page`	`int`	The page number. \| 页码。	required
`size`	`int`	The size of the page. \| 页的大小。	required
`type_include`	`Optional[list[str]]`	The message types to include. \| 要包含的消息类型。	`None`
`type_exclude`	`Optional[list[str]]`	The message types to exclude. \| 要排除的消息类型。	`None`
`role_include`	`Optional[list[str]]`	The message roles to include. \| 要包含的消息角色。	`None`
`role_exclude`	`Optional[list[str]]`	The message roles to exclude. \| 要排除的消息角色。	`None`
`filter_meta`	`Optional[Callable[[Attributes], bool]]`	The function to filter messages by metadata. \| 用于根据元数据过滤消息的函数。	`None`

Returns:

Type	Description
`list[UserAndAssMsg]`	list[UserAndAssMsg]: The messages from the conversation. \| 对话中的消息。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_conversation_messages(
    self,
    conversation_id: CONVERSATION_KEY,
    page: int,
    size: int,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[Callable[[Attributes], bool]] = None,
) -> list[UserAndAssMsg]:
    """
    Get messages from a conversation.

    Args:
        conversation_id (int | str): The id/index of the conversation. | 对话的索引。
        page (int): The page number. | 页码。
        size (int): The size of the page. | 页的大小。
        type_include (Optional[list[str]]): The message types to include. | 要包含的消息类型。
        type_exclude (Optional[list[str]]): The message types to exclude. | 要排除的消息类型。
        role_include (Optional[list[str]]): The message roles to include. | 要包含的消息角色。
        role_exclude (Optional[list[str]]): The message roles to exclude. | 要排除的消息角色。
        filter_meta (Optional[Callable[[Attributes], bool]]): The function to filter messages by metadata. | 用于根据
            元数据过滤消息的函数。

    Returns:
        list[UserAndAssMsg]: The messages from the conversation. | 对话中的消息。
    """
    ...

aget_conversation_messages `abstractmethod` `async` ¶

aget_conversation_messages(
    conversation_id: CONVERSATION_KEY,
    page: int,
    size: int,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[
        Callable[[Attributes], bool]
    ] = None,
) -> list[UserAndAssMsg]

Get messages from a conversation asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_conversation_messages(
    self,
    conversation_id: CONVERSATION_KEY,
    page: int,
    size: int,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[Callable[[Attributes], bool]] = None,
) -> list[UserAndAssMsg]:
    """Get messages from a conversation asynchronously."""
    ...

get_latest_msg_from_conversation `abstractmethod` ¶

get_latest_msg_from_conversation(
    conversation_id: CONVERSATION_KEY,
) -> Optional[UserAndAssMsg]

Get the latest message from a conversation.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_latest_msg_from_conversation(self, conversation_id: CONVERSATION_KEY) -> Optional[UserAndAssMsg]:
    """Get the latest message from a conversation."""
    ...

aget_latest_msg_from_conversation `abstractmethod` `async` ¶

aget_latest_msg_from_conversation(
    conversation_id: CONVERSATION_KEY,
) -> Optional[UserAndAssMsg]

Get the latest message from a conversation asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_latest_msg_from_conversation(self, conversation_id: CONVERSATION_KEY) -> Optional[UserAndAssMsg]:
    """Get the latest message from a conversation asynchronously."""
    ...

get_conversation_messages_by_cursor `abstractmethod` ¶

get_conversation_messages_by_cursor(
    conversation_id: CONVERSATION_KEY,
    count: int,
    cursor: Optional[str] = None,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[
        Callable[[Attributes], bool]
    ] = None,
) -> tuple[list[UserAndAssMsg], str]

Get messages from a conversation by cursor.

Count可以是负数，如果是负数，表示从后向前取。

Parameters:

Name	Type	Description	Default
`conversation_id`	`int \| str`	The id/index of the conversation. \| 对话的索引。	required
`cursor`	`str`	The cursor to get messages. \| 获取消息的游标。	`None`
`count`	`int`	The number of messages to get. \| 要获取的消息数量。	required
`type_include`	`Optional[list[str]]`	The message types to include. \| 要包含的消息类型。	`None`
`type_exclude`	`Optional[list[str]]`	The message types to exclude. \| 要排除的消息类型。	`None`
`role_include`	`Optional[list[str]]`	The message roles to include. \| 要包含的消息角色。	`None`
`role_exclude`	`Optional[list[str]]`	The message roles to exclude. \| 要排除的消息角色。	`None`
`filter_meta`	`Optional[Callable[[Attributes], bool]]`	The function to filter messages by metadata. \| 用于根据	`None`

Returns:

Type	Description
`tuple[list[UserAndAssMsg], str]`	tuple[list[UserAndAssMsg], str]: The messages from the conversation and the cursor. \| 对话中的消息和游标。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_conversation_messages_by_cursor(
    self,
    conversation_id: CONVERSATION_KEY,
    count: int,
    cursor: Optional[str] = None,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[Callable[[Attributes], bool]] = None,
) -> tuple[list[UserAndAssMsg], str]:
    """
    Get messages from a conversation by cursor.

    Count可以是负数，如果是负数，表示从后向前取。

    Args:
        conversation_id (int | str): The id/index of the conversation. | 对话的索引。
        cursor (str): The cursor to get messages. | 获取消息的游标。
        count (int): The number of messages to get. | 要获取的消息数量。
        type_include (Optional[list[str]]): The message types to include. | 要包含的消息类型。
        type_exclude (Optional[list[str]]): The message types to exclude. | 要排除的消息类型。
        role_include (Optional[list[str]]): The message roles to include. | 要包含的消息角色。
        role_exclude (Optional[list[str]]): The message roles to exclude. | 要排除的消息角色。
        filter_meta (Optional[Callable[[Attributes], bool]]): The function to filter messages by metadata. | 用于根据

    Returns:
        tuple[list[UserAndAssMsg], str]: The messages from the conversation and the cursor. | 对话中的消息和游标。
    """
    ...

aget_conversation_messages_by_cursor `abstractmethod` `async` ¶

aget_conversation_messages_by_cursor(
    conversation_id: CONVERSATION_KEY,
    count: int,
    cursor: Optional[str] = None,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[
        Callable[[Attributes], bool]
    ] = None,
) -> tuple[list[UserAndAssMsg], str]

Get messages from a conversation asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_conversation_messages_by_cursor(
    self,
    conversation_id: CONVERSATION_KEY,
    count: int,
    cursor: Optional[str] = None,
    type_include: Optional[list[str]] = None,
    type_exclude: Optional[list[str]] = None,
    role_include: Optional[list[str]] = None,
    role_exclude: Optional[list[str]] = None,
    filter_meta: Optional[Callable[[Attributes], bool]] = None,
) -> tuple[list[UserAndAssMsg], str]:
    """Get messages from a conversation asynchronously."""
    ...

add_conversation `abstractmethod` ¶

add_conversation(
    name: str, platform_id: Optional[PLATFORM_ID] = None
) -> CONVERSATION_KEY

Add a conversation to memory.

是否需要传递平台ID，取决不同的实现。但建议尽量传递。

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the conversation. \| 对话的名称。	required
`platform_id`	`Optional[int]`	The platform id of the conversation. \| 对话的平台ID。	`None`

Returns:

Type	Description
`CONVERSATION_KEY`	int \| str: The id of the added conversation. \| 添加的对话的ID。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_conversation(self, name: str, platform_id: Optional[PLATFORM_ID] = None) -> CONVERSATION_KEY:
    """
    Add a conversation to memory.

    是否需要传递平台ID，取决不同的实现。但建议尽量传递。

    Args:
        name (str): The name of the conversation. | 对话的名称。
        platform_id (Optional[int]): The platform id of the conversation. | 对话的平台ID。

    Returns:
        int | str: The id of the added conversation. | 添加的对话的ID。
    """
    ...

aadd_conversation `abstractmethod` `async` ¶

aadd_conversation(
    name: str, platform_id: Optional[PLATFORM_ID] = None
) -> CONVERSATION_KEY

Add a conversation asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_conversation(self, name: str, platform_id: Optional[PLATFORM_ID] = None) -> CONVERSATION_KEY:
    """Add a conversation asynchronously."""
    ...

update_conversation `abstractmethod` ¶

update_conversation(
    conversation_id: CONVERSATION_KEY, name: str
) -> None

Update a conversation in memory.

Parameters:

Name	Type	Description	Default
`conversation_id`	`int \| str`	The id/index of the conversation. \| 对话的索引。	required
`name`	`str`	The name of the conversation. \| 对话的名称。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_conversation(self, conversation_id: CONVERSATION_KEY, name: str) -> None:
    """
    Update a conversation in memory.

    Args:
        conversation_id (int | str): The id/index of the conversation. | 对话的索引。
        name (str): The name of the conversation. | 对话的名称。
    """
    ...

aupdate_conversation `abstractmethod` `async` ¶

aupdate_conversation(
    conversation_id: CONVERSATION_KEY, name: str
) -> None

Update a conversation asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_conversation(self, conversation_id: CONVERSATION_KEY, name: str) -> None:
    """Update a conversation asynchronously."""
    ...

delete_conversation `abstractmethod` ¶

delete_conversation(
    conversation_id: CONVERSATION_KEY,
) -> None

Delete a conversation from memory.

Parameters:

Name	Type	Description	Default
`conversation_id`	`int \| str`	The id/index of the conversation. \| 对话的索引。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_conversation(self, conversation_id: CONVERSATION_KEY) -> None:
    """
    Delete a conversation from memory.

    Args:
        conversation_id (int | str): The id/index of the conversation. | 对话的索引。
    """
    ...

adelete_conversation `abstractmethod` `async` ¶

adelete_conversation(
    conversation_id: CONVERSATION_KEY,
) -> None

Delete a conversation from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_conversation(self, conversation_id: CONVERSATION_KEY) -> None:
    """Delete a conversation from memory asynchronously."""
    ...

get_docs `abstractmethod` ¶

get_docs(
    page: Optional[int] = None,
    size: Optional[int] = None,
    keywords: Optional[list[str]] = None,
) -> list[Document]

Get all documents from memory.

从记忆中获取所有文档。

Parameters:

Name	Type	Description	Default
`page`	`Optional[int]`	The page number. \| 页码。	`None`
`size`	`Optional[int]`	The size of the page. \| 页的大小。	`None`
`keywords`	`Optional[list[str]]`	The keywords to search for. \| 要搜索的关键字	`None`

Returns:

Type	Description
`list[Document]`	list[DocElement]: The list of all documents.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_docs(
    self, page: Optional[int] = None, size: Optional[int] = None, keywords: Optional[list[str]] = None
) -> list[Document]:
    """
    Get all documents from memory.

    从记忆中获取所有文档。

    Args:
        page (Optional[int]): The page number. | 页码。
        size (Optional[int]): The size of the page. | 页的大小。
        keywords (Optional[list[str]]): The keywords to search for. | 要搜索的关键字

    Returns:
        list[DocElement]: The list of all documents.
    """
    ...

aget_docs `abstractmethod` `async` ¶

aget_docs(
    page: Optional[int] = None, size: Optional[int] = None
) -> list[Document]

Get all documents from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_docs(self, page: Optional[int] = None, size: Optional[int] = None) -> list[Document]:
    """Get all documents from memory asynchronously."""
    ...

get_doc `abstractmethod` ¶

get_doc(doc_id: int) -> Document

Get a document from memory.

从记忆中获取文档。

Parameters:

Name	Type	Description	Default
`doc_id`	`int`	The id of the document to get. \| 要获取的文档的ID。	required

Returns:

Name	Type	Description
`Document`	`Document`	The document. \| 文档。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_doc(self, doc_id: int) -> Document:
    """
    Get a document from memory.

    从记忆中获取文档。

    Args:
        doc_id (int): The id of the document to get. | 要获取的文档的ID。

    Returns:
        Document: The document. | 文档。
    """
    ...

aget_doc `abstractmethod` `async` ¶

aget_doc(doc_id: int) -> Document

Get a document from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_doc(self, doc_id: int) -> Document:
    """Get a document from memory asynchronously."""
    ...

add_doc `abstractmethod` ¶

add_doc(doc: Document) -> DocId

Add a document to memory.

添加文档到记忆中。

Parameters:

Name	Type	Description	Default
`doc`	`Document`	The document to add. \| 要添加的文档。	required

Returns:

Name	Type	Description
`DocId`	`DocId`	The id of the added document.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_doc(self, doc: Document) -> DocId:
    """
    Add a document to memory.

    添加文档到记忆中。

    Args:
        doc (Document): The document to add. | 要添加的文档。

    Returns:
        DocId: The id of the added document.
    """
    ...

aadd_doc `abstractmethod` `async` ¶

aadd_doc(doc: Document) -> DocId

Add a document asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_doc(self, doc: Document) -> DocId:
    """Add a document asynchronously."""
    ...

update_doc `abstractmethod` ¶

update_doc(doc: Document) -> None

Update a document in memory.

更新记忆中的文档。

Parameters:

Name	Type	Description	Default
`doc`	`Document`	The document to update. \| 要更新的文档。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_doc(self, doc: Document) -> None:
    """
    Update a document in memory.

    更新记忆中的文档。

    Args:
        doc (Document): The document to update. | 要更新的文档。
    """
    ...

aupdate_doc `abstractmethod` `async` ¶

aupdate_doc(doc: Document) -> None

Update a document in memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_doc(self, doc: Document) -> None:
    """Update a document in memory asynchronously."""
    ...

delete_doc `abstractmethod` ¶

delete_doc(doc_id: int) -> None

Delete a document from memory.

从记忆中删除文档。

Parameters:

Name	Type	Description	Default
`doc_id`	`int`	The id of the document to delete. \| 要删除的文档的ID。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_doc(self, doc_id: int) -> None:
    """
    Delete a document from memory.

    从记忆中删除文档。

    Args:
        doc_id (int): The id of the document to delete. | 要删除的文档的ID。
    """
    ...

adelete_doc `abstractmethod` `async` ¶

adelete_doc(doc_id: int) -> None

Delete a document from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_doc(self, doc_id: int) -> None:
    """Delete a document from memory asynchronously."""
    ...

get_pages `abstractmethod` ¶

get_pages(
    page: Optional[int],
    size: Optional[int],
    keywords: Optional[list[str]],
    doc_ids: Optional[list[int]],
) -> list[DocPage]

Get all pages from memory.

从记忆中获取所有页面。

Parameters:

Name	Type	Description	Default
`page`	`Optional[int]`	The page number.	required
`size`	`Optional[int]`	The size of the page.	required
`keywords`	`Optional[list[str]]`	The keywords to search for.	required
`doc_ids`	`Optional[list[int]]`	The document ids to search for.	required

Returns:

Type	Description
`list[DocPage]`	list[DocElement]: The list of all pages.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_pages(
    self, page: Optional[int], size: Optional[int], keywords: Optional[list[str]], doc_ids: Optional[list[int]]
) -> list[DocPage]:
    """
    Get all pages from memory.

    从记忆中获取所有页面。

    Args:
        page (Optional[int]): The page number.
        size (Optional[int]): The size of the page.
        keywords (Optional[list[str]]): The keywords to search for.
        doc_ids (Optional[list[int]]): The document ids to search for.

    Returns:
        list[DocElement]: The list of all pages.
    """
    ...

aget_pages `abstractmethod` `async` ¶

aget_pages(
    page: Optional[int],
    size: Optional[int],
    keywords: Optional[list[str]],
    doc_ids: Optional[list[int]],
) -> list[DocPage]

Get all pages from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_pages(
    self, page: Optional[int], size: Optional[int], keywords: Optional[list[str]], doc_ids: Optional[list[int]]
) -> list[DocPage]:
    """Get all pages from memory asynchronously."""
    ...

get_page `abstractmethod` ¶

get_page(page_id: int) -> DocPage

Get a page from memory.

从记忆中获取页面。

Parameters:

Name	Type	Description	Default
`page_id`	`int`	The id of the page to get. \| 要获取的页面的ID。	required

Returns:

Name	Type	Description
`DocPage`	`DocPage`	The page. \| 页面。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_page(self, page_id: int) -> DocPage:
    """
    Get a page from memory.

    从记忆中获取页面。

    Args:
        page_id (int): The id of the page to get. | 要获取的页面的ID。

    Returns:
        DocPage: The page. | 页面。
    """
    ...

aget_page `abstractmethod` `async` ¶

aget_page(page_id: int) -> DocPage

Get a page from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_page(self, page_id: int) -> DocPage:
    """Get a page from memory asynchronously."""
    ...

add_page `abstractmethod` ¶

add_page(page: DocPage) -> PageId

Add a page to memory.

添加页面到记忆中。

Parameters:

Name	Type	Description	Default
`page`	`DocPage`	The page to add. \| 要添加的页面。	required

Returns:

Name	Type	Description
`PageId`	`PageId`	The id of the added page.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_page(self, page: DocPage) -> PageId:
    """
    Add a page to memory.

    添加页面到记忆中。

    Args:
        page (DocPage): The page to add. | 要添加的页面。

    Returns:
        PageId: The id of the added page.
    """
    ...

aadd_page `abstractmethod` `async` ¶

aadd_page(page: DocPage) -> PageId

Add a page to memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_page(self, page: DocPage) -> PageId:
    """Add a page to memory asynchronously."""
    ...

update_page `abstractmethod` ¶

update_page(page: DocPage) -> None

Update a page in memory.

更新记忆中的页面。

Parameters:

Name	Type	Description	Default
`page`	`DocPage`	The page to update. \| 要更新的页面。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_page(self, page: DocPage) -> None:
    """
    Update a page in memory.

    更新记忆中的页面。

    Args:
        page (DocPage): The page to update. | 要更新的页面。
    """
    ...

aupdate_page `abstractmethod` `async` ¶

aupdate_page(page: DocPage) -> None

Update a page in memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_page(self, page: DocPage) -> None:
    """Update a page in memory asynchronously."""
    ...

delete_page `abstractmethod` ¶

delete_page(page_id: int) -> None

Delete a page from memory.

从记忆中删除页面。

Parameters:

Name	Type	Description	Default
`page_id`	`int`	The id of the page to delete. \| 要删除的页面的ID。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_page(self, page_id: int) -> None:
    """
    Delete a page from memory.

    从记忆中删除页面。

    Args:
        page_id (int): The id of the page to delete. | 要删除的页面的ID。
    """
    ...

adelete_page `abstractmethod` `async` ¶

adelete_page(page_id: int) -> None

Delete a page from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_page(self, page_id: int) -> None:
    """Delete a page from memory asynchronously."""
    ...

get_elements `abstractmethod` ¶

get_elements(
    page: Optional[int],
    size: Optional[int],
    keywords: Optional[list[str]],
    page_ids: Optional[list[int]],
) -> list[DocElement]

Get all elements from memory.

从记忆中获取所有元素。

Parameters:

Name	Type	Description	Default
`page`	`Optional[int]`	The page number.	required
`size`	`Optional[int]`	The size of the page.	required
`keywords`	`Optional[list[str]]`	The keywords to search for.	required
`page_ids`	`Optional[list[int]]`	The page ids to search for.	required

Returns:

Type	Description
`list[DocElement]`	list[DocElement]: The list of all elements.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_elements(
    self, page: Optional[int], size: Optional[int], keywords: Optional[list[str]], page_ids: Optional[list[int]]
) -> list[DocElement]:
    """
    Get all elements from memory.

    从记忆中获取所有元素。

    Args:
        page (Optional[int]): The page number.
        size (Optional[int]): The size of the page.
        keywords (Optional[list[str]]): The keywords to search for.
        page_ids (Optional[list[int]]): The page ids to search for.

    Returns:
        list[DocElement]: The list of all elements.
    """
    ...

aget_elements `abstractmethod` `async` ¶

aget_elements(
    page: Optional[int],
    size: Optional[int],
    keywords: Optional[list[str]],
    page_ids: Optional[list[int]],
) -> list[DocElement]

Get all elements from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_elements(
    self, page: Optional[int], size: Optional[int], keywords: Optional[list[str]], page_ids: Optional[list[int]]
) -> list[DocElement]:
    """Get all elements from memory asynchronously."""
    ...

get_element `abstractmethod` ¶

get_element(element_id: int) -> DocElement

Get an element from memory.

从记忆中获取元素。

Parameters:

Name	Type	Description	Default
`element_id`	`int`	The id of the element to get. \| 要获取文档元素的ID。	required

Returns:

Name	Type	Description
`DocElement`	`DocElement`	The element. \| 元素。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_element(self, element_id: int) -> DocElement:
    """
    Get an element from memory.

    从记忆中获取元素。

    Args:
        element_id (int): The id of the element to get. | 要获取文档元素的ID。

    Returns:
        DocElement: The element. | 元素。
    """
    ...

aget_element `abstractmethod` `async` ¶

aget_element(element_id: int) -> DocElement

Get an element from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_element(self, element_id: int) -> DocElement:
    """Get an element from memory asynchronously."""
    ...

add_element `abstractmethod` ¶

add_element(element: DocElement) -> EleId

Add an element to memory.

添加元素到记忆中。

Parameters:

Name	Type	Description	Default
`element`	`DocElement`	The element to add. \| 要添加的元素。	required

Returns:

Type	Description
`EleId`	EleId

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_element(self, element: DocElement) -> EleId:
    """
    Add an element to memory.

    添加元素到记忆中。

    Args:
        element (DocElement): The element to add. |  要添加的元素。

    Returns:
        EleId
    """
    ...

aadd_element `abstractmethod` `async` ¶

aadd_element(element: DocElement) -> EleId

Add an element from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_element(self, element: DocElement) -> EleId:
    """Add an element from memory asynchronously."""
    ...

update_element `abstractmethod` ¶

update_element(element: DocElement) -> None

Update an element in memory.

更新记忆中的元素。

Parameters:

Name	Type	Description	Default
`element`	`DocElement`	The element to update. \| 要更新的元素。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_element(self, element: DocElement) -> None:
    """
    Update an element in memory.

    更新记忆中的元素。

    Args:
        element (DocElement): The element to update. | 要更新的元素。
    """
    ...

aupdate_element `abstractmethod` `async` ¶

aupdate_element(element: DocElement) -> None

Update an element in memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_element(self, element: DocElement) -> None:
    """Update an element in memory asynchronously."""
    ...

delete_element `abstractmethod` ¶

delete_element(element_id: int) -> None

Delete an element from memory.

从记忆中删除元素。

Parameters:

Name	Type	Description	Default
`element_id`	`int`	The id of the element to delete. \| 要删除文档元素的ID。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_element(self, element_id: int) -> None:
    """
    Delete an element from memory.

    从记忆中删除元素。

    Args:
        element_id (int): The id of the element to delete. | 要删除文档元素的ID。
    """
    ...

adelete_element `abstractmethod` `async` ¶

adelete_element(element_id: int) -> None

Delete an element from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_element(self, element_id: int) -> None:
    """Delete an element from memory asynchronously."""
    ...

get_graph_cls `abstractmethod` ¶

get_graph_cls(class_iri: CLS_IRI) -> Optional[ThingClass]

Get a graph class from memory.

Parameters:

Name	Type	Description	Default
`class_iri`	`CLS_IRI`	The class iri. \| 类的IRI。	required

Returns:

Type	Description
`Optional[ThingClass]`	Optional[ThingClass]: The class information

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_graph_cls(self, class_iri: CLS_IRI) -> Optional[ThingClass]:
    """
    Get a graph class from memory.

    Args:
        class_iri (CLS_IRI): The class iri. | 类的IRI。

    Returns:
        Optional[ThingClass]: The class information
    """
    ...

aget_graph_cls `abstractmethod` `async` ¶

aget_graph_cls(class_iri: CLS_IRI) -> ThingClass

Get a graph class from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_graph_cls(self, class_iri: CLS_IRI) -> ThingClass:
    """Get a graph class from memory asynchronously."""
    ...

get_all_graph_clses `abstractmethod` ¶

get_all_graph_clses() -> list[ThingClass]

Get all graph classes from memory.

从记忆中获取所有图类。

Returns:

Type	Description
`list[ThingClass]`	list[ThingClass]: The list of all graph classes. \| 所有图类的列表。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_all_graph_clses(self) -> list[ThingClass]:
    """
    Get all graph classes from memory.

    从记忆中获取所有图类。

    Returns:
        list[ThingClass]: The list of all graph classes. | 所有图类的列表。
    """
    ...

aget_all_graph_clses `abstractmethod` `async` ¶

aget_all_graph_clses() -> list[ThingClass]

Get all graph classes from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_all_graph_clses(self) -> list[ThingClass]:
    """Get all graph classes from memory asynchronously."""
    ...

add_graph_cls `abstractmethod` ¶

add_graph_cls(
    class_iri: CLS_IRI,
    super_classes: Optional[list[str]] = None,
    annotations: Optional[dict] = None,
) -> CLS_IRI

Add a graph class to memory.

Parameters:

Name	Type	Description	Default
`class_iri`	`CLS_IRI`	The class iri. \| 类的IRI。	required
`super_classes`	`Optional[list[str]]`	The super classes of the class. \| 类的超类。	`None`
`annotations`	`Optional[dict]`	The annotations of the class. \| 类的注解。	`None`

Returns:

Name	Type	Description
`CLS_IRI`	`CLS_IRI`	The class iri. \| 类的IRI。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_graph_cls(
    self, class_iri: CLS_IRI, super_classes: Optional[list[str]] = None, annotations: Optional[dict] = None
) -> CLS_IRI:
    """
    Add a graph class to memory.

    Args:
        class_iri (CLS_IRI): The class iri. | 类的IRI。
        super_classes (Optional[list[str]]): The super classes of the class. | 类的超类。
        annotations (Optional[dict]): The annotations of the class. | 类的注解。

    Returns:
        CLS_IRI: The class iri. | 类的IRI。
    """
    ...

aadd_graph_cls `abstractmethod` `async` ¶

aadd_graph_cls(
    class_iri: CLS_IRI,
    super_classes: Optional[list[str]] = None,
    annotations: Optional[dict] = None,
) -> CLS_IRI

Add a graph class from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_graph_cls(
    self, class_iri: CLS_IRI, super_classes: Optional[list[str]] = None, annotations: Optional[dict] = None
) -> CLS_IRI:
    """Add a graph class from memory asynchronously."""
    ...

update_graph_cls `abstractmethod` ¶

update_graph_cls(
    class_iri: CLS_IRI,
    new_super_classes: Optional[list[str]] = None,
    new_annotations: Optional[dict] = None,
) -> None

Update a graph class in memory

需要注意new_annotations 会对原属性先清空后添加，如果需要删除，则使用空列表或者空值即可。

另外需要注意，更新时 pos 与 name_properties 需要全部提供，建议使用类似http中put的更新方式(即提供所有信息)，而不是patch的方式(即只提供需要更新的信息)。

Parameters:

Name	Type	Description	Default
`class_iri`	`CLS_IRI`	The class iri. \| 类的IRI。	required
`new_super_classes`	`Optional[list[str]]`	The new super classes of the class. \| 类的新超类。	`None`
`new_annotations`	`Optional[dict]`	The new annotations of the class. \| 类的新注解。	`None`

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_graph_cls(
    self, class_iri: CLS_IRI, new_super_classes: Optional[list[str]] = None, new_annotations: Optional[dict] = None
) -> None:
    """
    Update a graph class in memory

    需要注意new_annotations 会对原属性先清空后添加，如果需要删除，则使用空列表或者空值即可。

    另外需要注意，更新时 pos 与 name_properties 需要全部提供，建议使用类似http中put的更新方式(即提供所有信息)，而不是patch的方式(即只提供需要更新的信息)。

    Args:
        class_iri (CLS_IRI):  The class iri. | 类的IRI。
        new_super_classes (Optional[list[str]]): The new super classes of the class. | 类的新超类。
        new_annotations (Optional[dict]): The new annotations of the class. | 类的新注解。
    """
    ...

aupdate_graph_cls `abstractmethod` `async` ¶

aupdate_graph_cls(
    class_iri: CLS_IRI,
    new_super_classes: Optional[list[str]] = None,
    new_annotations: Optional[dict] = None,
) -> None

Update a graph class from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_graph_cls(
    self, class_iri: CLS_IRI, new_super_classes: Optional[list[str]] = None, new_annotations: Optional[dict] = None
) -> None:
    """Update a graph class from memory asynchronously."""
    ...

delete_graph_cls `abstractmethod` ¶

delete_graph_cls(class_iri: CLS_IRI) -> None

Delete a graph class from memory.

Parameters:

Name	Type	Description	Default
`class_iri`	`CLS_IRI`	The class iri. \| 类的IRI。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_graph_cls(self, class_iri: CLS_IRI) -> None:
    """
    Delete a graph class from memory.

    Args:
        class_iri (CLS_IRI): The class iri. | 类的IRI。
    """
    ...

adelete_graph_cls `abstractmethod` `async` ¶

adelete_graph_cls(class_iri: CLS_IRI) -> None

Delete a graph class from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_graph_cls(self, class_iri: CLS_IRI) -> None:
    """Delete a graph class from memory asynchronously."""
    ...

get_graph_property `abstractmethod` ¶

get_graph_property(
    property_iri: PROP_IRI,
) -> Optional[PropertyClass]

Get a graph property from memory.

Parameters:

Name	Type	Description	Default
`property_iri`	`PROP_IRI`	The property iri. \| 属性的IRI。	required

Returns:

Type	Description
`Optional[PropertyClass]`	Optional[PropertyClass]: The property information. \| 属性的信息。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_graph_property(self, property_iri: PROP_IRI) -> Optional[PropertyClass]:
    """
    Get a graph property from memory.

    Args:
        property_iri (PROP_IRI): The property iri. | 属性的IRI。

    Returns:
        Optional[PropertyClass]: The property information. | 属性的信息。
    """
    ...

aget_graph_property `abstractmethod` `async` ¶

aget_graph_property(
    property_iri: PROP_IRI,
) -> PropertyClass

Get a graph property from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_graph_property(self, property_iri: PROP_IRI) -> PropertyClass:
    """Get a graph property from memory asynchronously."""
    ...

get_all_graph_properties `abstractmethod` ¶

get_all_graph_properties(
    prop_type: Optional[
        Literal["object", "data", "annotation"]
    ] = None,
) -> list[PropertyClass]

Get all graph properties from memory.

从记忆中获取所有图属性。

Parameters:

Name	Type	Description	Default
`prop_type`	`Optional[Literal['object', 'data', 'annotation']]`	The type of the property. \| 属性类型	`None`

Returns:

Type	Description
`list[PropertyClass]`	list[PropertyClass]: The list of all graph properties. \| 所有图属性的列表。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_all_graph_properties(
    self, prop_type: Optional[Literal["object", "data", "annotation"]] = None
) -> list[PropertyClass]:
    """
    Get all graph properties from memory.

    从记忆中获取所有图属性。

    Args:
        prop_type (Optional[Literal["object", "data", "annotation"]]): The type of the property. | 属性类型

    Returns:
        list[PropertyClass]: The list of all graph properties. | 所有图属性的列表。
    """
    ...

aget_all_graph_properties `abstractmethod` `async` ¶

aget_all_graph_properties(
    prop_type: Optional[
        Literal["object", "data", "annotation"]
    ] = None,
) -> list[PropertyClass]

Get all graph properties from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_all_graph_properties(
    self, prop_type: Optional[Literal["object", "data", "annotation"]] = None
) -> list[PropertyClass]:
    """Get all graph properties from memory asynchronously."""
    ...

add_graph_property `abstractmethod` ¶

add_graph_property(
    property_iri: PROP_IRI,
    property_type: Literal["object", "data"],
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: bool = False,
    is_inverse_functional: bool = False,
    is_symmetric: bool = False,
    is_transitive: bool = False,
    is_asymmetric: bool = False,
    is_reflexive: bool = False,
    is_irreflexive: bool = False,
    trigger_words: Optional[list[str]] = None,
) -> PROP_IRI

Add a graph property to memory.

Parameters:

Name	Type	Description	Default
`property_iri`	`PROP_IRI`	The property iri. \| 属性的IRI。	required
`property_type`	`Literal['object', 'data']`	The property type. \| 属性类型。	required
`domain`	`Optional[list[CLS_IRI]]`	The domain of the property. \| 属性的域。	`None`
`o_range`	`Optional[list]`	The range of the property. \| 属性的范围。	`None`
`is_functional`	`bool`	Whether the property is functional. \| 属性是否功能性。	`False`
`is_inverse_functional`	`bool`	Whether the property is inverse functional. \| 属性是否反功能性。	`False`
`is_symmetric`	`bool`	Whether the property is symmetric. \| 属性是否对称。	`False`
`is_transitive`	`bool`	Whether the property is transitive. \| 属性是否传递。	`False`
`is_asymmetric`	`bool`	Whether the property is asymmetric. \| 属性是否反对称。	`False`
`is_reflexive`	`bool`	Whether the property is reflexive. \| 属性是否自反。	`False`
`is_irreflexive`	`bool`	Whether the property is irreflexive. \| 属性是否非自反。	`False`
`trigger_words`	`Optional[list[str]]`	The trigger words of the property. \| 属性的触发词。	`None`

Returns:

Name	Type	Description
`PROP_IRI`	`PROP_IRI`	The property iri. \| 属性的IRI。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_graph_property(
    self,
    property_iri: PROP_IRI,
    property_type: Literal["object", "data"],
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: bool = False,
    is_inverse_functional: bool = False,
    is_symmetric: bool = False,
    is_transitive: bool = False,
    is_asymmetric: bool = False,
    is_reflexive: bool = False,
    is_irreflexive: bool = False,
    trigger_words: Optional[list[str]] = None,
) -> PROP_IRI:
    """
    Add a graph property to memory.

    Args:
        property_iri (PROP_IRI): The property iri. | 属性的IRI。
        property_type (Literal["object", "data"]): The property type. | 属性类型。
        domain (Optional[list[CLS_IRI]]): The domain of the property. | 属性的域。
        o_range (Optional[list]): The range of the property. | 属性的范围。
        is_functional (bool): Whether the property is functional. | 属性是否功能性。
        is_inverse_functional (bool): Whether the property is inverse functional. | 属性是否反功能性。
        is_symmetric (bool): Whether the property is symmetric. | 属性是否对称。
        is_transitive (bool): Whether the property is transitive. | 属性是否传递。
        is_asymmetric (bool): Whether the property is asymmetric. | 属性是否反对称。
        is_reflexive (bool): Whether the property is reflexive. | 属性是否自反。
        is_irreflexive (bool): Whether the property is irreflexive. | 属性是否非自反。
        trigger_words (Optional[list[str]]): The trigger words of the property. | 属性的触发词。

    Returns:
        PROP_IRI: The property iri. | 属性的IRI。
    """
    ...

aadd_graph_property `abstractmethod` `async` ¶

aadd_graph_property(
    property_iri: PROP_IRI,
    property_type: Literal["object", "data"],
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: bool = False,
    is_inverse_functional: bool = False,
    is_symmetric: bool = False,
    is_transitive: bool = False,
    is_asymmetric: bool = False,
    is_reflexive: bool = False,
    is_irreflexive: bool = False,
    trigger_words: Optional[list[str]] = None,
) -> PROP_IRI

Add a graph property from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_graph_property(
    self,
    property_iri: PROP_IRI,
    property_type: Literal["object", "data"],
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: bool = False,
    is_inverse_functional: bool = False,
    is_symmetric: bool = False,
    is_transitive: bool = False,
    is_asymmetric: bool = False,
    is_reflexive: bool = False,
    is_irreflexive: bool = False,
    trigger_words: Optional[list[str]] = None,
) -> PROP_IRI:
    """Add a graph property from memory asynchronously."""
    ...

update_graph_property `abstractmethod` ¶

update_graph_property(
    property_iri: PROP_IRI,
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: Optional[bool] = None,
    is_inverse_functional: Optional[bool] = None,
    is_symmetric: Optional[bool] = None,
    is_transitive: Optional[bool] = None,
    is_asymmetric: Optional[bool] = None,
    is_reflexive: Optional[bool] = None,
    is_irreflexive: Optional[bool] = None,
    trigger_words: Optional[list[str]] = None,
) -> None

Update a graph property in memory.

Parameters:

Name	Type	Description	Default
`property_iri`	`PROP_IRI`	The property iri. \| 属性的IRI。	required
`domain`	`Optional[list[CLS_IRI]]`	The new domain of the property. \| 新属性的域。	`None`
`o_range`	`Optional[list]`	The new range of the property. \| 新属性的范围。	`None`
`is_functional`	`Optional[bool]`	Whether the property is functional. \| 新属性是否功能性。	`None`
`is_inverse_functional`	`Optional[bool]`	Whether the property is inverse functional. \| 新属性是否反功能性。	`None`
`is_symmetric`	`Optional[bool]`	Whether the property is symmetric. \| 新属性是否对称。	`None`
`is_transitive`	`Optional[bool]`	Whether	`None`
`is_asymmetric`	`Optional[bool]`	Whether the property is asymmetric. \| 新属性是否反对称。	`None`
`is_reflexive`	`Optional[bool]`	Whether the property is reflexive. \| 新属性��否自反。	`None`
`is_irreflexive`	`Optional[bool]`	Whether the property is irreflexive. \| 新属性是否非自反。	`None`
`trigger_words`	`Optional[list[str]]`	The new trigger words of the property. \| 新属性的触发词。	`None`

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_graph_property(
    self,
    property_iri: PROP_IRI,
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: Optional[bool] = None,
    is_inverse_functional: Optional[bool] = None,
    is_symmetric: Optional[bool] = None,
    is_transitive: Optional[bool] = None,
    is_asymmetric: Optional[bool] = None,
    is_reflexive: Optional[bool] = None,
    is_irreflexive: Optional[bool] = None,
    trigger_words: Optional[list[str]] = None,
) -> None:
    """
    Update a graph property in memory.

    Args:
        property_iri (PROP_IRI): The property iri. | 属性的IRI。
        domain (Optional[list[CLS_IRI]]): The new domain of the property. | 新属性的域。
        o_range (Optional[list]): The new range of the property. | 新属性的范围。
        is_functional (Optional[bool]): Whether the property is functional. | 新属性是否功能性。
        is_inverse_functional (Optional[bool]): Whether the property is inverse functional. | 新属性是否反功能性。
        is_symmetric (Optional[bool]): Whether the property is symmetric. | 新属性是否对称。
        is_transitive (Optional[bool]): Whether
        is_asymmetric (Optional[bool]): Whether the property is asymmetric. | 新属性是否反对称。
        is_reflexive (Optional[bool]): Whether the property is reflexive. | 新属性���否自反。
        is_irreflexive (Optional[bool]): Whether the property is irreflexive. | 新属性是否非自反。
        trigger_words (Optional[list[str]]): The new trigger words of the property. | 新属性的触发词。
    """
    ...

aupdate_graph_property `abstractmethod` `async` ¶

aupdate_graph_property(
    property_iri: PROP_IRI,
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: Optional[bool] = None,
    is_inverse_functional: Optional[bool] = None,
    is_symmetric: Optional[bool] = None,
    is_transitive: Optional[bool] = None,
    is_asymmetric: Optional[bool] = None,
    is_reflexive: Optional[bool] = None,
    is_irreflexive: Optional[bool] = None,
    trigger_words: Optional[list[str]] = None,
) -> None

Update a graph property in memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_graph_property(
    self,
    property_iri: PROP_IRI,
    domain: Optional[list[CLS_IRI]] = None,
    o_range: Optional[list] = None,
    is_functional: Optional[bool] = None,
    is_inverse_functional: Optional[bool] = None,
    is_symmetric: Optional[bool] = None,
    is_transitive: Optional[bool] = None,
    is_asymmetric: Optional[bool] = None,
    is_reflexive: Optional[bool] = None,
    is_irreflexive: Optional[bool] = None,
    trigger_words: Optional[list[str]] = None,
) -> None:
    """Update a graph property in memory asynchronously."""
    ...

get_graph_entity `abstractmethod` ¶

get_graph_entity(entity_iri: IND_IRI) -> Optional[Thing]

Get a graph entity from memory.

Parameters:

Name	Type	Description	Default
`entity_iri`	`IND_IRI`	The entity iri. \| 实体的IRI。	required

Returns:

Name	Type	Description
`Thing`	`Optional[Thing]`	The entity

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_graph_entity(self, entity_iri: IND_IRI) -> Optional[Thing]:
    """
    Get a graph entity from memory.

    Args:
        entity_iri (IND_IRI): The entity iri. | 实体的IRI。

    Returns:
        Thing: The entity
    """
    ...

aget_graph_entity `abstractmethod` `async` ¶

aget_graph_entity(entity_iri: IND_IRI) -> Thing

Get a graph entity from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_graph_entity(self, entity_iri: IND_IRI) -> Thing:
    """Get a graph entity from memory asynchronously."""
    ...

get_graph_entities_by_cls `abstractmethod` ¶

get_graph_entities_by_cls(cls_iri: CLS_IRI) -> list[Thing]

Get all graph entities from memory.

从记忆中获取所有图实体。

Parameters:

Name	Type	Description	Default
`cls_iri`	`CLS_IRI`	The class iri. \| 类的IRI。	required

Returns:

Type	Description
`list[Thing]`	list[Thing]: The list of all graph entities. \| 所有图实体的列表。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_graph_entities_by_cls(self, cls_iri: CLS_IRI) -> list[Thing]:
    """
    Get all graph entities from memory.

    从记忆中获取所有图实体。

    Args:
        cls_iri (CLS_IRI): The class iri. | 类的IRI。

    Returns:
        list[Thing]: The list of all graph entities. | 所有图实体的列表。
    """
    ...

aget_graph_entities_by_cls `abstractmethod` `async` ¶

aget_graph_entities_by_cls(cls_iri: CLS_IRI) -> Thing

Get all graph entities from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_graph_entities_by_cls(self, cls_iri: CLS_IRI) -> Thing:
    """Get all graph entities from memory asynchronously."""
    ...

get_all_graph_entities `abstractmethod` ¶

get_all_graph_entities() -> list[Thing]

Get all graph entities from memory.

从记忆中获取所有图实体。

Returns:

Type	Description
`list[Thing]`	list[Thing]: The list of all graph entities. \| 所有图实体的列表。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def get_all_graph_entities(self) -> list[Thing]:
    """
    Get all graph entities from memory.

    从记忆中获取所有图实体。

    Returns:
        list[Thing]: The list of all graph entities. | 所有图实体的列表。
    """
    ...

aget_all_graph_entities `abstractmethod` `async` ¶

aget_all_graph_entities() -> list[Thing]

Get all graph entities from memory.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aget_all_graph_entities(self) -> list[Thing]:
    """Get all graph entities from memory."""
    ...

add_graph_entity `abstractmethod` ¶

add_graph_entity(cls_iri: CLS_IRI, info: dict) -> IND_IRI

Add a graph entity to memory.

Parameters:

Name	Type	Description	Default
`cls_iri`	`str`	The entity iri. \| 实体的IRI。	required
`info`	`str`	The entity information. \| 实体的信息。	required

Returns:

Name	Type	Description
`IND_IRI`	`IND_IRI`	The entity iri. \| 实体的IRI。

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def add_graph_entity(self, cls_iri: CLS_IRI, info: dict) -> IND_IRI:
    """
    Add a graph entity to memory.

    Args:
        cls_iri (str): The entity iri. | 实体的IRI。
        info (str): The entity information. | 实体的信息。

    Returns:
        IND_IRI: The entity iri. | 实体的IRI。
    """
    ...

aadd_graph_entity `abstractmethod` `async` ¶

aadd_graph_entity(cls_iri: CLS_IRI, info: dict) -> IND_IRI

Add a graph entity to memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aadd_graph_entity(self, cls_iri: CLS_IRI, info: dict) -> IND_IRI:
    """Add a graph entity to memory asynchronously."""
    ...

update_graph_entity `abstractmethod` ¶

update_graph_entity(
    entity_iri: IND_IRI, info: dict
) -> None

Update a graph entity in memory.

Parameters:

Name	Type	Description	Default
`entity_iri`	`IND_IRI`	The entity iri. \| 实体的IRI。	required
`info`	`str`	The entity information. \| 实体的信息。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def update_graph_entity(self, entity_iri: IND_IRI, info: dict) -> None:
    """
    Update a graph entity in memory.

    Args:
        entity_iri (IND_IRI): The entity iri. | 实体的IRI。
        info (str): The entity information. | 实体的信息。
    """
    ...

aupdate_graph_entity `abstractmethod` `async` ¶

aupdate_graph_entity(
    entity_iri: IND_IRI, info: dict
) -> None

Update a graph entity in memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def aupdate_graph_entity(self, entity_iri: IND_IRI, info: dict) -> None:
    """Update a graph entity in memory asynchronously."""
    ...

delete_graph_entity `abstractmethod` ¶

delete_graph_entity(entity_iri: IND_IRI) -> None

Delete a graph entity from memory.

Parameters:

Name	Type	Description	Default
`entity_iri`	`IND_IRI`	The entity iri. \| 实体的IRI。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_graph_entity(self, entity_iri: IND_IRI) -> None:
    """
    Delete a graph entity from memory.

    Args:
        entity_iri (IND_IRI): The entity iri. | 实体的IRI。
    """
    ...

adelete_graph_entity `abstractmethod` `async` ¶

adelete_graph_entity(entity_iri: IND_IRI) -> None

Delete a graph entity from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_graph_entity(self, entity_iri: IND_IRI) -> None:
    """Delete a graph entity from memory asynchronously."""
    ...

delete_graph_property `abstractmethod` ¶

delete_graph_property(property_iri: PROP_IRI) -> None

Delete a graph property from memory.

Parameters:

Name	Type	Description	Default
`property_iri`	`PROP_IRI`	The property iri. \| 属性的IRI。	required

Source code in tfrobot/brain/memory/base.py

@abstractmethod
def delete_graph_property(self, property_iri: PROP_IRI) -> None:
    """
    Delete a graph property from memory.

    Args:
        property_iri (PROP_IRI): The property iri. | 属性的IRI。
    """
    ...

adelete_graph_property `abstractmethod` `async` ¶

adelete_graph_property(property_iri: PROP_IRI) -> None

Delete a graph property from memory asynchronously.

Source code in tfrobot/brain/memory/base.py

@abstractmethod
async def adelete_graph_property(self, property_iri: PROP_IRI) -> None:
    """Delete a graph property from memory asynchronously."""
    ...

get_save_config `classmethod` ¶

get_save_config() -> Tuple[BaseSaverConfig, set]

Generates configuration for save_to_dir function.

As all messages need to be stored in one file, using 'is_independent_storage' for split storage is not feasible because it would result in a large IO burden as each message would generate a file. Therefore, we use 'mode="json"' for serializing and deserializing, and incorporate it into the main file.

生成用于 save_to_dir 函数的配置信息。因为所有的消息都需要存储在一个文件中，所以使用 'is_independent_storage' 进行分裂存储是不可行的，因为这将导致每条消息生成一个文件，造成大的 IO 负担。所以我们使用 'mode="json"' 进行序列化和反序列化，并将其纳入主文件。

Returns:

Name	Type	Description
`Dict`	`Tuple[BaseSaverConfig, set]`	A dictionary containing the generated configuration. 包含生成的配置的字典。

Source code in tfrobot/brain/memory/base.py

@classmethod
def get_save_config(cls) -> Tuple[BaseSaverConfig, set]:
    """
    Generates configuration for save_to_dir function.

    As all messages need to be stored in one file, using 'is_independent_storage' for split storage is not feasible
    because it would result in a large IO burden as each message would generate a file.
    Therefore, we use 'mode="json"' for serializing and deserializing, and incorporate it into the main file.

    生成用于 save_to_dir 函数的配置信息。因为所有的消息都需要存储在一个文件中，所以使用 'is_independent_storage' 进行分裂存储是不可行的，
    因为这将导致每条消息生成一个文件，造成大的 IO 负担。所以我们使用 'mode="json"' 进行序列化和反序列化，并将其纳入主文件。

    Returns:
        Dict: A dictionary containing the generated configuration.

            包含生成的配置的字典。
    """
    # 从字段定义的元数据中找到tf_meta.is_independent_storage为True的字段名称，形成一个set
    independent_storage_attrs = set()
    for name, field in cls.model_fields.items():
        if isinstance(field.json_schema_extra, dict) and field.json_schema_extra.get("is_independent_storage"):
            independent_storage_attrs.add(name)
    return BaseSaverConfig(mode="json"), independent_storage_attrs

recall_in_neural ¶

recall_in_neural(
    sender: str,
    query: str,
    chunk_size: int | Annotated[list[int], Len(3, 3)],
    length_function: Callable[[str], int],
    exclude_str: Optional[str] = None,
) -> Tuple[
    Optional[list[UserAndAssMsg]],
    Optional[list[DocElement]],
    Optional[str],
]

Recalls the content from memory based on the given natural language input.

根据给定的自然语言输入从内存中检索内容。

Parameters:

Name	Type	Description	Default
`sender`	`str`	The sender of the message. 消息的发送者。	required
`query`	`str`	The natural language input. This is used to recall the content from memory. It is usually formatted by chat input history. 自然语言输入。用这个来从内存中回忆内容。通常由聊天输入历史记录格式化。	required
`chunk_size`	`Union[int, List[int]]`	The size of the content to recall. It can be a single number or a list of numbers. If it is a list, the first number is the size of the conversation content to recall, the second number is the size of the memo content to recall, and the third number is the size of the knowledge content to recall. If it is a single number, it will be converted into three equal numbers. 要回忆的内容的大小。可以是单个数字或数字列表。如果是列表，第一个数字是要回忆的对话内容的大小，第二个数字是要回忆的备忘内容的大小，第三个数字是要回忆的知识内容的大小。如果是单个数字，将被转换为三个相等的数字。	required
`length_function`	`Callable`	A function to calculate the length of the content to recall. 用于计算所回忆内容长度的函数。	required
`exclude_str`	`Optional[str]`	The string to exclude from the recall. `从回忆中排除的字符串。`	`None`

Returns:

Name	Type	Description
`Any`	`Tuple[Optional[list[UserAndAssMsg]], Optional[list[DocElement]], Optional[str]]`	The content recalled from memory. 从内存中回忆的内容。

Source code in tfrobot/brain/memory/base.py

def recall_in_neural(
    self,
    sender: str,
    query: str,
    chunk_size: int | Annotated[list[int], annotated_types.Len(3, 3)],  # type: ignore
    length_function: Callable[[str], int],
    exclude_str: Optional[str] = None,
) -> Tuple[Optional[list[UserAndAssMsg]], Optional[list[DocElement]], Optional[str]]:
    """
    Recalls the content from memory based on the given natural language input.

    根据给定的自然语言输入从内存中检索内容。

    Args:
        sender (str): The sender of the message.

            消息的发送者。
        query (str): The natural language input. This is used to recall the content from memory. It is usually
            formatted by chat input history.

            自然语言输入。用这个来从内存中回忆内容。通常由聊天输入历史记录格式化。

        chunk_size (Union[int, List[int]]): The size of the content to recall. It can be a single number or a list
            of numbers. If it is a list, the first number is the size of the conversation content to recall, the
            second number is the size of the memo content to recall, and the third number is the size of the
            knowledge content to recall. If it is a single number, it will be converted into three equal numbers.

            要回忆的内容的大小。可以是单个数字或数字列表。如果是列表，第一个数字是要回忆的
            对话内容的大小，第二个数字是要回忆的备忘内容的大小，第三个数字是要回忆的知识内容的大小。如果是单个数字，将被转换为三个相等的数字。

        length_function (Callable): A function to calculate the length of the content to recall.

            用于计算所回忆内容长度的函数。

        exclude_str (Optional[str]): The string to exclude from the recall.

                从回忆中排除的字符串。

    Returns:
        Any: The content recalled from memory.

            从内存中回忆的内容。
    """
    if sender != "memory_recall":
        raise ValueError("The sender is not the same as the registered sender.")
    query_msg: TextMessage = TextMessage(
        content=query, creator=BaseUser(uid="NOT_EXIST", name="NEURAL_INSTINCTIVE_RESPONSE")
    )
    return self.recall(query_msg, chunk_size, length_function, exclude_str=exclude_str)

dispose ¶

dispose() -> None

释放记忆体持有的资源。基类 no-op，子类按需覆写。

Source code in tfrobot/brain/memory/base.py

def dispose(self) -> None:
    """释放记忆体持有的资源。基类 no-op，子类按需覆写。"""
    pass

adispose `async` ¶

adispose() -> None

异步释放记忆体持有的资源。基类 no-op，子类按需覆写。

Source code in tfrobot/brain/memory/base.py

async def adispose(self) -> None:
    """异步释放记忆体持有的资源。基类 no-op，子类按需覆写。"""
    pass

connect_to_neural ¶

connect_to_neural(neural: Neural) -> None

实现NeuralProtocol协议，向Neural注册自己

Parameters:

Name	Type	Description	Default
`neural`	`Neural`	Neural实例	required

Returns:

Type	Description
`None`	None

Source code in tfrobot/brain/memory/base.py

def connect_to_neural(self, neural: Neural) -> None:
    """
    实现NeuralProtocol协议，向Neural注册自己

    Args:
        neural (Neural): Neural实例

    Returns:
        None
    """
    if self._neural and self._neural is not neural:
        raise ValueError("The memory is already connected to another neural.")
    self._neural = neural
    self._neural.signal.connect(self.recall_in_neural, "memory_recall")

disconnect_from_neural ¶

disconnect_from_neural(neural: Neural) -> None

实现NeuralProtocol协议，从Neural注销自己

Parameters:

Name	Type	Description	Default
`neural`	`Neural`	Neural实例	required

Returns:

Type	Description
`None`	None

Source code in tfrobot/brain/memory/base.py

def disconnect_from_neural(self, neural: Neural) -> None:
    """
    实现NeuralProtocol协议，从Neural注销自己

    Args:
        neural (Neural): Neural实例

    Returns:
        None
    """
    if self._neural is not neural:  # pragma: no cover
        raise ValueError("The neural is not the same as the registered neural.")  # pragma: no cover
    self._neural.signal.disconnect(self.recall_in_neural, "memory_recall")  # pragma: no cover
    self._neural = None  # pragma: no cover

记忆体基类 (BaseMemory)¶

BaseMemory ¶

recall abstractmethod ¶

async_recall abstractmethod async ¶

commit abstractmethod ¶

acommit abstractmethod async ¶

sync_msg ¶

async_sync_msg async ¶

get_platforms abstractmethod ¶

aget_platforms abstractmethod async ¶

add_platform abstractmethod ¶

aadd_platform abstractmethod async ¶

update_platform abstractmethod ¶

aupdate_platform abstractmethod async ¶

delete_platform abstractmethod ¶

adelete_platform abstractmethod async ¶

get_conversations abstractmethod ¶

aget_conversations abstractmethod async ¶

get_conversation abstractmethod ¶

aget_conversation abstractmethod async ¶

get_conversation_messages abstractmethod ¶

aget_conversation_messages abstractmethod async ¶

get_latest_msg_from_conversation abstractmethod ¶

aget_latest_msg_from_conversation abstractmethod async ¶

get_conversation_messages_by_cursor abstractmethod ¶

aget_conversation_messages_by_cursor abstractmethod async ¶

add_conversation abstractmethod ¶

aadd_conversation abstractmethod async ¶

update_conversation abstractmethod ¶

aupdate_conversation abstractmethod async ¶

delete_conversation abstractmethod ¶

adelete_conversation abstractmethod async ¶

get_docs abstractmethod ¶

aget_docs abstractmethod async ¶

get_doc abstractmethod ¶

aget_doc abstractmethod async ¶

add_doc abstractmethod ¶

aadd_doc abstractmethod async ¶

update_doc abstractmethod ¶

aupdate_doc abstractmethod async ¶

delete_doc abstractmethod ¶

adelete_doc abstractmethod async ¶

get_pages abstractmethod ¶

aget_pages abstractmethod async ¶

get_page abstractmethod ¶

aget_page abstractmethod async ¶

add_page abstractmethod ¶

aadd_page abstractmethod async ¶

update_page abstractmethod ¶

aupdate_page abstractmethod async ¶

delete_page abstractmethod ¶

adelete_page abstractmethod async ¶

get_elements abstractmethod ¶

aget_elements abstractmethod async ¶

get_element abstractmethod ¶

aget_element abstractmethod async ¶

add_element abstractmethod ¶

aadd_element abstractmethod async ¶

update_element abstractmethod ¶

aupdate_element abstractmethod async ¶

delete_element abstractmethod ¶

adelete_element abstractmethod async ¶

get_graph_cls abstractmethod ¶

aget_graph_cls abstractmethod async ¶

get_all_graph_clses abstractmethod ¶

aget_all_graph_clses abstractmethod async ¶

add_graph_cls abstractmethod ¶

aadd_graph_cls abstractmethod async ¶

update_graph_cls abstractmethod ¶

aupdate_graph_cls abstractmethod async ¶

delete_graph_cls abstractmethod ¶

adelete_graph_cls abstractmethod async ¶

get_graph_property abstractmethod ¶

aget_graph_property abstractmethod async ¶

get_all_graph_properties abstractmethod ¶

aget_all_graph_properties abstractmethod async ¶

add_graph_property abstractmethod ¶

aadd_graph_property abstractmethod async ¶

update_graph_property abstractmethod ¶

aupdate_graph_property abstractmethod async ¶

recall `abstractmethod` ¶

async_recall `abstractmethod` `async` ¶

commit `abstractmethod` ¶

acommit `abstractmethod` `async` ¶

async_sync_msg `async` ¶

get_platforms `abstractmethod` ¶

aget_platforms `abstractmethod` `async` ¶

add_platform `abstractmethod` ¶

aadd_platform `abstractmethod` `async` ¶

update_platform `abstractmethod` ¶

aupdate_platform `abstractmethod` `async` ¶

delete_platform `abstractmethod` ¶

adelete_platform `abstractmethod` `async` ¶

get_conversations `abstractmethod` ¶

aget_conversations `abstractmethod` `async` ¶

get_conversation `abstractmethod` ¶

aget_conversation `abstractmethod` `async` ¶

get_conversation_messages `abstractmethod` ¶

aget_conversation_messages `abstractmethod` `async` ¶

get_latest_msg_from_conversation `abstractmethod` ¶

aget_latest_msg_from_conversation `abstractmethod` `async` ¶

get_conversation_messages_by_cursor `abstractmethod` ¶

aget_conversation_messages_by_cursor `abstractmethod` `async` ¶

add_conversation `abstractmethod` ¶

aadd_conversation `abstractmethod` `async` ¶

update_conversation `abstractmethod` ¶

aupdate_conversation `abstractmethod` `async` ¶

delete_conversation `abstractmethod` ¶

adelete_conversation `abstractmethod` `async` ¶

get_docs `abstractmethod` ¶

aget_docs `abstractmethod` `async` ¶

get_doc `abstractmethod` ¶

aget_doc `abstractmethod` `async` ¶

add_doc `abstractmethod` ¶

aadd_doc `abstractmethod` `async` ¶

update_doc `abstractmethod` ¶

aupdate_doc `abstractmethod` `async` ¶

delete_doc `abstractmethod` ¶

adelete_doc `abstractmethod` `async` ¶

get_pages `abstractmethod` ¶

aget_pages `abstractmethod` `async` ¶

get_page `abstractmethod` ¶

aget_page `abstractmethod` `async` ¶

add_page `abstractmethod` ¶

aadd_page `abstractmethod` `async` ¶

update_page `abstractmethod` ¶

aupdate_page `abstractmethod` `async` ¶

delete_page `abstractmethod` ¶

adelete_page `abstractmethod` `async` ¶

get_elements `abstractmethod` ¶

aget_elements `abstractmethod` `async` ¶

get_element `abstractmethod` ¶

aget_element `abstractmethod` `async` ¶

add_element `abstractmethod` ¶

aadd_element `abstractmethod` `async` ¶

update_element `abstractmethod` ¶

aupdate_element `abstractmethod` `async` ¶

delete_element `abstractmethod` ¶

adelete_element `abstractmethod` `async` ¶

get_graph_cls `abstractmethod` ¶

aget_graph_cls `abstractmethod` `async` ¶

get_all_graph_clses `abstractmethod` ¶

aget_all_graph_clses `abstractmethod` `async` ¶

add_graph_cls `abstractmethod` ¶

aadd_graph_cls `abstractmethod` `async` ¶

update_graph_cls `abstractmethod` ¶

aupdate_graph_cls `abstractmethod` `async` ¶

delete_graph_cls `abstractmethod` ¶

adelete_graph_cls `abstractmethod` `async` ¶

get_graph_property `abstractmethod` ¶

aget_graph_property `abstractmethod` `async` ¶

get_all_graph_properties `abstractmethod` ¶

aget_all_graph_properties `abstractmethod` `async` ¶

add_graph_property `abstractmethod` ¶

aadd_graph_property `abstractmethod` `async` ¶

update_graph_property `abstractmethod` ¶

aupdate_graph_property `abstractmethod` `async` ¶

get_graph_entity `abstractmethod` ¶

aget_graph_entity `abstractmethod` `async` ¶

get_graph_entities_by_cls `abstractmethod` ¶