agentscope.memory

The memory module.

class MemoryBase[源代码]

基类:StateModule

The base class for memory in agentscope.

abstract async add(*args, **kwargs)[源代码]

Add items to the memory.

参数:
  • args (Any)

  • kwargs (Any)

返回类型:

None

abstract async delete(*args, **kwargs)[源代码]

Delete items from the memory.

参数:
  • args (Any)

  • kwargs (Any)

返回类型:

None

abstract async retrieve(*args, **kwargs)[源代码]

Retrieve items from the memory.

参数:
  • args (Any)

  • kwargs (Any)

返回类型:

None

abstract async size()[源代码]

Get the size of the memory.

返回类型:

int

abstract async clear()[源代码]

Clear the memory content.

返回类型:

None

abstract async get_memory(*args, **kwargs)[源代码]

Get the memory content.

参数:
  • args (Any)

  • kwargs (Any)

返回类型:

list[Msg]

abstract state_dict()[源代码]

Get the state dictionary of the memory.

返回类型:

dict

abstract load_state_dict(state_dict, strict=True)[源代码]

Load the state dictionary of the memory.

参数:
  • state_dict (dict)

  • strict (bool)

返回类型:

None

class InMemoryMemory[源代码]

基类:MemoryBase

The in-memory memory class for storing messages.

__init__()[源代码]

Initialize the in-memory memory object.

返回类型:

None

state_dict()[源代码]

Convert the current memory into JSON data format.

返回类型:

dict

load_state_dict(state_dict, strict=True)[源代码]

Load the memory from JSON data.

参数:
  • state_dict (dict) -- The state dictionary to load, which should have a "content" field.

  • strict (bool, defaults to True) -- If True, raises an error if any key in the module is not found in the state_dict. If False, skips missing keys.

返回类型:

None

async size()[源代码]

The size of the memory.

返回类型:

int

async retrieve(*args, **kwargs)[源代码]

Retrieve items from the memory.

参数:
  • args (Any)

  • kwargs (Any)

返回类型:

None

async delete(index)[源代码]

Delete the specified item by index(es).

参数:

index (Union[Iterable, int]) -- The index to delete.

返回类型:

None

async add(memories, allow_duplicates=False)[源代码]

Add message into the memory.

参数:
  • memories (Union[list[Msg], Msg, None]) -- The message to add.

  • allow_duplicates (bool, defaults to False) -- If allow adding duplicate messages (with the same id) into the memory.

返回类型:

None

async get_memory()[源代码]

Get the memory content.

返回类型:

list[Msg]

async clear()[源代码]

Clear the memory content.

返回类型:

None

class LongTermMemoryBase[源代码]

基类:StateModule

The long-term memory base class, which should be a time-series memory management system.

The record_to_memory and retrieve_from_memory methods are two tool functions for agent to manage the long-term memory voluntarily. You can choose not to implement these two functions.

The record and retrieve methods are for developers to use. For example, retrieving/recording memory at the beginning of each reply, and adding the retrieved memory to the system prompt.

async record(msgs, **kwargs)[源代码]

A developer-designed method to record information from the given input message(s) to the long-term memory.

参数:
  • msgs (list[Msg | None])

  • kwargs (Any)

返回类型:

None

async retrieve(msg, **kwargs)[源代码]

A developer-designed method to retrieve information from the long-term memory based on the given input message(s). The retrieved information will be added to the system prompt of the agent.

参数:
  • msg (Msg | list[Msg] | None)

  • kwargs (Any)

返回类型:

str

async record_to_memory(thinking, content, **kwargs)[源代码]

Use this function to record important information that you may need later. The target content should be specific and concise, e.g. who, when, where, do what, why, how, etc.

参数:
  • thinking (str) -- Your thinking and reasoning about what to record

  • content (list[str]) -- The content to remember, which is a list of strings.

  • kwargs (Any)

返回类型:

ToolResponse

async retrieve_from_memory(keywords, **kwargs)[源代码]

Retrieve the memory based on the given keywords.

参数:
  • keywords (list[str]) -- The keywords to search for in the memory, which should be specific and concise, e.g. the person's name, the date, the location, etc.

  • kwargs (Any)

返回:

A list of messages that match the keywords.

返回类型:

list[Msg]

class Mem0LongTermMemory[源代码]

基类:LongTermMemoryBase

A class that implements the LongTermMemoryBase interface using mem0.

__init__(agent_name=None, user_name=None, run_name=None, model=None, embedding_model=None, vector_store_config=None, mem0_config=None, default_memory_type=None, **kwargs)[源代码]

Initialize the Mem0LongTermMemory instance

参数:
  • agent_name (str | None, optional) -- The name of the agent. Default is None.

  • user_name (str | None, optional) -- The name of the user. Default is None.

  • run_name (str | None, optional) -- The name of the run/session. Default is None.

  • model (ChatModelBase | None)

  • embedding_model (EmbeddingModelBase | None)

  • vector_store_config (Any | None)

  • mem0_config (Any | None)

  • default_memory_type (str | None)

  • kwargs (Any)

返回类型:

None

备注

  1. At least one of agent_name, user_name, or run_name is required.

  2. During memory recording, these parameters become metadata for the stored memories.

  3. Important: mem0 will extract memories from messages containing role of "user" by default. If you want to extract memories from messages containing role of "assistant", you need to provide agent_name.

  4. During memory retrieval, only memories with matching metadata values will be returned.

model (ChatModelBase | None, optional):

The chat model to use for the long-term memory. If mem0_config is provided, this will override the LLM configuration. If mem0_config is None, this is required.

embedding_model (EmbeddingModelBase | None, optional):

The embedding model to use for the long-term memory. If mem0_config is provided, this will override the embedder configuration. If mem0_config is None, this is required.

vector_store_config (VectorStoreConfig | None, optional):

The vector store config to use for the long-term memory. If mem0_config is provided, this will override the vector store configuration. If mem0_config is None and this is not provided, defaults to Qdrant with on_disk=True.

mem0_config (MemoryConfig | None, optional):

The mem0 config to use for the long-term memory. If provided, individual model/embedding_model/vector_store_config parameters will override the corresponding configurations in mem0_config. If None, a new MemoryConfig will be created using the provided parameters.

default_memory_type (str | None, optional):

The type of memory to use. Default is None, to create a semantic memory.

抛出:

ValueError -- If mem0_config is None and either model or embedding_model is None.

参数:
  • agent_name (str | None)

  • user_name (str | None)

  • run_name (str | None)

  • model (ChatModelBase | None)

  • embedding_model (EmbeddingModelBase | None)

  • vector_store_config (Any | None)

  • mem0_config (Any | None)

  • default_memory_type (str | None)

  • kwargs (Any)

返回类型:

None

async record_to_memory(thinking, content, **kwargs)[源代码]

Use this function to record important information that you may need later. The target content should be specific and concise, e.g. who, when, where, do what, why, how, etc.

参数:
  • thinking (str) -- Your thinking and reasoning about what to record.

  • content (list[str]) -- The content to remember, which is a list of strings.

  • kwargs (Any)

返回类型:

ToolResponse

async retrieve_from_memory(keywords, limit=5, **kwargs)[源代码]

Retrieve the memory based on the given keywords.

参数:
  • keywords (list[str]) -- The keywords to search for in the memory, which should be specific and concise, e.g. the person's name, the date, the location, etc.

  • limit (int, optional) -- The maximum number of memories to retrieve per search.

  • kwargs (Any)

返回:

A ToolResponse containing the retrieved memories as JSON text.

返回类型:

ToolResponse

async record(msgs, memory_type=None, infer=True, **kwargs)[源代码]

Record the content to the long-term memory.

参数:
  • msgs (list[Msg | None]) -- The messages to record to memory.

  • memory_type (str | None, optional) -- The type of memory to use. Default is None, to create a semantic memory. "procedural_memory" is explicitly used for procedural memories.

  • infer (bool, optional) -- Whether to infer memory from the content. Default is True.

  • **kwargs (Any) -- Additional keyword arguments for the mem0 recording.

返回类型:

None

async retrieve(msg, limit=5, **kwargs)[源代码]

Retrieve the content from the long-term memory.

参数:
  • msg (Msg | list[Msg] | None) -- The message to search for in the memory, which should be specific and concise, e.g. the person's name, the date, the location, etc.

  • limit (int, optional) -- The maximum number of memories to retrieve per search.

  • **kwargs (Any) -- Additional keyword arguments.

返回:

The retrieved memory

返回类型:

str

class ReMePersonalLongTermMemory[源代码]

基类:ReMeLongTermMemoryBase

Personal memory implementation using ReMe library.

Requirements:

Python 3.12 or greater is required to use ReMe.

async record_to_memory(thinking, content, **kwargs)[源代码]

Record important user information to long-term memory.

Record important user information to long-term memory for future reference.

Use this function to save user's personal information, preferences, habits, and facts that you may need in future conversations. This enables you to provide personalized and contextually relevant responses.

When to record:

  • User shares personal preferences (e.g., "I prefer homestays when traveling")

  • User mentions habits or routines (e.g., "I start work at 9 AM")

  • User states likes/dislikes (e.g., "I enjoy drinking green tea")

  • User provides personal facts (e.g., "I work as a software engineer")

What to record: Be specific and structured. Include who, when, where, what, why, and how when relevant.

参数:
  • thinking (str) -- Your reasoning about why this information is worth recording and how it might be useful later.

  • content (list[str]) -- List of specific facts to remember. Each string should be a clear, standalone piece of information. Examples: ["User prefers homestays in Hangzhou", "User likes visiting West Lake in the morning"].

  • **kwargs (Any) -- Additional keyword arguments for the recording operation.

返回:

Confirmation message indicating successful memory recording.

返回类型:

ToolResponse

async retrieve_from_memory(keywords, **kwargs)[源代码]

Search and retrieve relevant information from long-term memory.

备注

You should call this function BEFORE answering questions about the user's preferences, past information, or personal details. This ensures you provide accurate information based on stored memories rather than guessing.

Use this when:

  • User asks "what do I like?", "what are my preferences?", "what do you know about me?"

  • User asks about their past behaviors, habits, or stated preferences

  • User refers to information they shared in previous conversations

  • You need to personalize responses based on user's history

参数:
  • keywords (list[str]) -- Keywords to search for in memory. Be specific and use multiple keywords for better results. Examples: ["travel preferences", "Hangzhou"], ["work habits", "morning routine"], ["food preferences", "tea"].

  • **kwargs (Any) -- Additional keyword arguments for the retrieval operation.

返回:

Retrieved memories matching the keywords. If no memories found, you'll receive a message indicating that.

返回类型:

ToolResponse

async record(msgs, **kwargs)[源代码]

Record the content to the long-term memory.

This method converts AgentScope messages to ReMe's format and records them using the personal memory flow.

参数:
  • msgs (list[Msg | None]) -- The messages to record to memory.

  • **kwargs (Any) -- Additional keyword arguments for the mem0 recording.

返回类型:

None

async retrieve(msg, **kwargs)[源代码]

Retrieve the content from the long-term memory.

参数:
  • msg (Msg | list[Msg] | None) -- The message to search for in the memory, which should be specific and concise, e.g. the person's name, the date, the location, etc.

  • **kwargs (Any) -- Additional keyword arguments.

返回:

The retrieved memory as a string.

返回类型:

str

class ReMeTaskLongTermMemory[源代码]

基类:ReMeLongTermMemoryBase

Task memory implementation using ReMe library.

Task memory learns from execution trajectories and provides retrieval of relevant task experiences.

Requirements:

Python 3.12 or greater is required to use ReMe.

async record_to_memory(thinking, content, **kwargs)[源代码]

Record task execution experiences and learnings.

Record task execution experiences and learnings to long-term memory.

Use this function to save valuable task-related knowledge that can help with future similar tasks. This enables learning from experience and improving over time.

When to record:

  • After solving technical problems or completing tasks

  • When discovering useful techniques or approaches

  • After implementing solutions with specific steps

  • When learning best practices or important lessons

What to record: Be detailed and actionable. Include:

  • Task description and context

  • Step-by-step execution details

  • Specific techniques and methods used

  • Results, outcomes, and effectiveness

  • Lessons learned and considerations

参数:
  • thinking (str) -- Your reasoning about why this task experience is valuable and what makes it worth remembering for future reference.

  • content (list[str]) -- List of specific task insights to remember. Each string should be a clear, actionable piece of information. Examples: ["Add indexes on WHERE clause columns to speed up queries", "Use EXPLAIN ANALYZE to identify missing indexes"].

  • **kwargs (Any) -- Additional keyword arguments. Can include 'score' (float) to indicate the quality/success of this approach (default: 1.0).

返回:

Confirmation message indicating successful memory recording.

返回类型:

ToolResponse

async retrieve_from_memory(keywords, **kwargs)[源代码]

Search and retrieve relevant task experiences.

Search and retrieve relevant task experiences from long-term memory.

IMPORTANT: You should call this function BEFORE attempting to solve problems or answer technical questions. This ensures you leverage experiences and proven solutions rather than starting from scratch.

Use this when: - Asked to solve a technical problem or implement a solution - Asked for recommendations, best practices, or approaches - Asked "what do you know about...?" or "have you seen this

before?"

  • Dealing with tasks that may be similar to experiences

  • Need to recall specific techniques or methods

Benefits of retrieving first: - Learn from past successes and mistakes - Provide more accurate, battle-tested solutions - Avoid reinventing the wheel - Give consistent, informed recommendations

参数:
  • keywords (list[str]) -- Keywords describing the task or problem domain. Be specific and use technical terms. Examples: ["database optimization", "slow queries"], ["API design", "rate limiting"], ["code refactoring", "Python"].

  • **kwargs (Any) -- Additional keyword arguments. Can include 'top_k' (int) to specify number of experiences to retrieve (default: 3).

返回:

Retrieved task experiences and learnings. If no relevant experiences found, you'll receive a message indicating that.

返回类型:

ToolResponse

async record(msgs, **kwargs)[源代码]

Record the content to the task memory.

This method converts AgentScope messages to ReMe's format and records them as a task execution trajectory.

参数:
  • msgs (list[Msg | None]) -- The messages to record to memory.

  • **kwargs (Any) -- Additional keyword arguments for the recording. Can include 'score' (float) for trajectory scoring (default: 1.0).

返回类型:

None

async retrieve(msg, **kwargs)[源代码]

Retrieve relevant task experiences from memory.

参数:
  • msg (Msg | list[Msg] | None) -- The message to search for relevant task experiences.

  • **kwargs (Any) -- Additional keyword arguments.

返回:

The retrieved task experiences as a string.

返回类型:

str

class ReMeToolLongTermMemory[源代码]

基类:ReMeLongTermMemoryBase

Tool memory implementation using ReMe library.

Tool memory records tool execution results and generates usage guidelines from the execution history.

Requirements:

Python 3.12 or greater is required to use ReMe.

async record_to_memory(thinking, content, **kwargs)[源代码]

Record tool execution results to build tool usage patterns.

Record tool execution results to build a knowledge base of tool usage patterns.

Use this function after successfully using tools to capture execution details, results, and performance metrics. Over time, this builds comprehensive usage guidelines and best practices for each tool.

When to record:

  • After successfully executing any tool

  • After tool failures (to learn what doesn't work)

  • When discovering effective parameter combinations

  • After noteworthy tool usage patterns

What to record: Each tool execution should include complete execution details.

参数:
  • thinking (str) -- Your reasoning about why this tool execution is worth recording. Mention what worked well, what could be improved, or lessons learned.

  • content (list[str]) --

    List of JSON strings, each representing a tool execution. Each JSON must have these fields: - create_time: Timestamp in format "YYYY-MM-DD HH:MM:SS" - tool_name: Name of the tool executed - input: Input parameters as a dict - output: Tool's output as a string - token_cost: Token cost (integer) - success: Whether execution succeeded (boolean) - time_cost: Execution time in seconds (float)

    Example: '{"create_time": "2024-01-01 10:00:00", "tool_name": "search", "input": {"query": "Python"}, "output": "Found 10 results", "token_cost": 100, "success": true, "time_cost": 1.2}'

  • **kwargs (Any) -- Additional keyword arguments for the recording operation.

返回:

Confirmation message with number of executions recorded and guidelines generated.

返回类型:

ToolResponse

async retrieve_from_memory(keywords, **kwargs)[源代码]

Retrieve usage guidelines and best practices for tools.

Retrieve usage guidelines and best practices for specific tools.

备注

You should call this function BEFORE using a tool, especially if you're uncertain about its proper usage or want to follow established best practices. This retrieves synthesized guidelines based on past tool executions.

Use this when:

  • About to use a tool and want to know the best practices

  • Uncertain about tool parameters or usage patterns

  • Want to learn from past successful/failed tool executions

  • User asks "how should I use this tool?" or "what's the best way to..."

  • Need to understand tool performance characteristics or limitations

Benefits of retrieving first:

  • Learn from accumulated tool usage experience

  • Avoid common mistakes and pitfalls

  • Use optimal parameter combinations

  • Understand tool performance and cost characteristics

  • Follow established best practices

参数:
  • keywords (list[str]) -- List of tool names to retrieve guidelines for. Use the exact tool names. Examples: ["search"], ["database_query", "cache_get"], ["api_call"].

  • **kwargs (Any) -- Additional keyword arguments for the retrieval operation.

返回:

Retrieved usage guidelines and best practices for the specified tools. If no guidelines exist yet, you'll receive a message indicating that.

返回类型:

ToolResponse

async record(msgs, **kwargs)[源代码]

Record the content to the tool memory.

This method extracts content from messages and treats them as JSON strings representing tool_call_results, similar to record_to_memory.

参数:
  • msgs (list[Msg | None]) -- The messages to record to memory. Each message's content should be a JSON string or list of JSON strings representing tool_call_results.

  • **kwargs (Any) -- Additional keyword arguments for the recording.

返回类型:

None

async retrieve(msg, **kwargs)[源代码]

Retrieve tool guidelines from memory.

Retrieve tool guidelines from memory based on message content.

参数:
  • msg (Msg | list[Msg] | None) -- The message containing tool names or queries to retrieve guidelines for.

  • **kwargs (Any) -- Additional keyword arguments.

返回:

The retrieved tool guidelines as a string.

返回类型:

str