Retrieve a specific memory by ID. Args: memory_id: The ID of the memory to retrieve. Returns: A dictionary c

Get the global Redis client instance.

Set default value for key if not exists.

Execute the underlying task with given arguments.

Update a memory by ID. Args: memory_id (str): ID of the memory to update. data (dict): Data to update the me

Count tokens in text.

Fetch a user by ID (with cache - Redis or IPS Cache).

Create an LLM client based on the model endpoint type. Args: llm_config: Configuration for the LLM model put

returns the project logger, scoped to a child name if provided Args: name: will define a child logger

Generate a string representation of the memory in-context using the Jinja2 template

Delete a textual memory from a MemCube by memory_id. Args: mem_cube_id (str): The identifier of the MemCube to delete th

Process log records of INFO or ERROR level (non-blocking)

* Feed raw data from stdout

Get all textual memories from a MemCube. Args: mem_cube_id (str, optional): The identifier of the MemCube to get the mem

Create a new client if it doesn't already exist (with caching).

Search raw memories with filtering, sorting, cursor-based pagination, and time range filtering. Args: organization_id: O

Make an HTTP request to the API. Args: method: HTTP method (GET, POST, etc.) endpoint: API endpoint (e.g., "

Stop all workers and close queue connections gracefully. Note: No logging during cleanup to avoid errors when logging system

Clear the entire graph.

Return a list of the blocks held inside the memory object

Dump the MemCube to a dictionary. Args: load_dir (str): The directory to load the MemCube from. user_id (str, optional

Close the database engine and dispose of all connections. This method should be called when the UserManager is no longer needed to en

* Called after new chunks are ingested. * Determines if a new task boundary was crossed and handles transition.

Detect the language of the given text (Chinese or English). Args: text: Text to analyze Returns: "zh" for Chinese, "en"

Fetch an agent by its ID (with cache and Redis pipeline for tools/blocks).

Add token usage from a single request.

Fetch an agent by its ID.

Get the active cache provider. Returns None if no provider is registered (graceful fallback to PostgreSQL). Returns: Cache prov

Retrieve node IDs based on vector similarity. Args: vector (list[float]): The embedding vector representing query semant

Test that a valid configuration can be instantiated. Args: cls: The config class to test valid_config: Dictionary of valid c

Create a new user. Args: user_name (str): Name of the user. role (UserRole): Role of the user. user_id (s

Apply filter_tags conditions and scope authorization to a SQLAlchemy ORM query. Args: query: SQLAlchemy Select statement to augment.

Remove Redis-specific fields before Pydantic validation. SECURITY: We keep Pydantic's extra="forbid" for API validation security.

Run the tool in a sandbox environment. Args: agent_state (Optional[AgentState]): The state of the agent invoking the too

Prepare query parameters for API requests. Args: kwargs: Keyword arguments to include in the parameters. Returns:

Add a memory node to the graph. Args: id: Unique identifier for the memory node. memory: Raw memory content (

Test that invalid configurations raise the appropriate exceptions. Args: cls: The config class to test invalid_config: Dicti

Fetch a single raw memory record by ID (with Redis JSON caching). Args: memory_id: ID of the memory to fetch

Initialize the queue and start the background workers. This method is idempotent — calling it multiple times only initializes once.

Load dataset.

Return LlamaIndex embedding model to use for embeddings

Factory function to get the appropriate graph client based on the graph type.