memory

memory

Redis memory services for ADK.

RedisLongTermMemoryService

RedisLongTermMemoryService(config: RedisLongTermMemoryServiceConfig | None = None)

Bases: BaseMemoryService

Long-term memory service backed by Redis memory backends.

The service implements ADK's memory interface using either Redis Agent Memory or the self-hosted Agent Memory Server. Searches are scoped by user ID and namespace for both backends.

Latest ADK versions add explicit add_events_to_memory and add_memory hooks. This service implements them while staying compatible with older ADK versions that only call add_session_to_memory and search_memory.

Initialize the Redis long-term memory service.

Parameters:

Name	Type	Description	Default
config	RedisLongTermMemoryServiceConfig | None	Configuration for the service. If None, defaults are used.	None

Source code in src/adk_redis/memory/long_term_memory.py

def __init__(self, config: RedisLongTermMemoryServiceConfig | None = None):
  """Initialize the Redis long-term memory service.

  Args:
      config: Configuration for the service. If None, defaults are used.
  """
  self._config = config or RedisLongTermMemoryServiceConfig()

add_session_to_memory async

add_session_to_memory(session: Session) -> None

Add an ADK session to the configured long-term memory backend.

Parameters:

Name	Type	Description	Default
session	Session	The ADK Session containing events to store.	required

Source code in src/adk_redis/memory/long_term_memory.py

@override
async def add_session_to_memory(self, session: Session) -> None:
  """Add an ADK session to the configured long-term memory backend.

  Args:
      session: The ADK Session containing events to store.
  """
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    try:
      await self._add_session_to_agent_memory_server(session)
    except Exception as e:
      logger.error(
          "Failed to add session %s to memory: %s",
          session.id,
          e,
      )
    return

  await self.add_events_to_memory(
      app_name=session.app_name,
      user_id=session.user_id,
      events=session.events,
      session_id=session.id,
  )

add_events_to_memory async

add_events_to_memory(*, app_name: str, user_id: str, events: Sequence[Event], session_id: str | None = None, custom_metadata: Mapping[str, object] | None = None) -> None

Add ADK events to the configured memory backend.

Source code in src/adk_redis/memory/long_term_memory.py

async def add_events_to_memory(
    self,
    *,
    app_name: str,
    user_id: str,
    events: Sequence[Event],
    session_id: str | None = None,
    custom_metadata: Mapping[str, object] | None = None,
) -> None:
  """Add ADK events to the configured memory backend."""
  if not self._config.store_events_as_messages:
    logger.debug("Event to memory storage is disabled")
    return

  try:
    if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
      session = Session(
          id=session_id
          or stable_memory_id(
              self._config.source,
              "events",
              app_name,
              user_id,
              *(getattr(event, "id", "") for event in events),
              *(getattr(event, "timestamp", "") for event in events),
          ),
          app_name=app_name,
          user_id=user_id,
          events=list(events),
      )
      await self._add_session_to_agent_memory_server(
          session,
          custom_metadata,
      )
      return

    records = self._build_event_records(
        app_name=app_name,
        user_id=user_id,
        events=events,
        session_id=session_id,
        custom_metadata=custom_metadata,
    )
    await self._bulk_create(records)
    logger.info("Stored %d event memory records", len(records))
  except Exception as e:
    logger.error("Failed to add events to memory: %s", e)

add_memory async

add_memory(*, app_name: str, user_id: str, memories: Sequence[MemoryEntry], custom_metadata: Mapping[str, object] | None = None) -> None

Add explicit durable memories to the configured backend.

Source code in src/adk_redis/memory/long_term_memory.py

async def add_memory(
    self,
    *,
    app_name: str,
    user_id: str,
    memories: Sequence[MemoryEntry],
    custom_metadata: Mapping[str, object] | None = None,
) -> None:
  """Add explicit durable memories to the configured backend."""
  try:
    if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
      await self._add_memory_to_agent_memory_server(
          app_name=app_name,
          user_id=user_id,
          memories=memories,
          custom_metadata=custom_metadata,
      )
      logger.info("Stored %d explicit memory records", len(memories))
      return

    records = self._build_memory_records(
        app_name=app_name,
        user_id=user_id,
        memories=memories,
        custom_metadata=custom_metadata,
    )
    await self._bulk_create(records)
    logger.info("Stored %d explicit memory records", len(records))
  except Exception as e:
    logger.error("Failed to add explicit memories: %s", e)

search_memory async

search_memory(*, app_name: str, user_id: str, query: str) -> SearchMemoryResponse

Search for memories using the configured memory backend.

Parameters:

Name	Type	Description	Default
app_name	str	Application name, used as namespace if no default is set.	required
user_id	str	Owner ID used to isolate memories.	required
query	str	Search query for semantic matching.	required

Returns:

Type	Description
SearchMemoryResponse	SearchMemoryResponse containing matching MemoryEntry objects.

Source code in src/adk_redis/memory/long_term_memory.py

@override
async def search_memory(
    self, *, app_name: str, user_id: str, query: str
) -> SearchMemoryResponse:
  """Search for memories using the configured memory backend.

  Args:
      app_name: Application name, used as namespace if no default is set.
      user_id: Owner ID used to isolate memories.
      query: Search query for semantic matching.

  Returns:
      SearchMemoryResponse containing matching MemoryEntry objects.
  """
  try:
    if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
      return await self._search_agent_memory_server(
          app_name=app_name,
          user_id=user_id,
          query=query,
      )

    namespace = self._get_namespace(app_name)
    request: dict[str, Any] = {
        "text": query,
        "limit": self._config.search_top_k,
        "filter": {
            "ownerId": {"eq": user_id},
            "namespace": {"eq": namespace},
        },
        "filterOp": "all",
    }
    threshold = self._search_threshold()
    if threshold is not None:
      request["similarityThreshold"] = threshold

    async with self._agent_memory() as agent_memory:
      results = await agent_memory.search_long_term_memory_async(
          request=request
      )

    memories = []
    for record in self._coerce_memories(results):
      text = str(read_field(record, "text", "") or "")
      if not text:
        continue

      created_at = read_field(record, "created_at")
      timestamp = created_at.isoformat() if created_at else None
      memory_type = self._memory_type_value(read_field(record, "memory_type"))
      content = types.Content(parts=[types.Part(text=text)])
      memories.append(
          MemoryEntry(
              id=read_field(record, "id"),
              timestamp=timestamp,
              content=content,
              custom_metadata={
                  "namespace": read_field(record, "namespace"),
                  "owner_id": read_field(record, "owner_id"),
                  "session_id": read_field(record, "session_id"),
                  "topics": read_field(record, "topics", []) or [],
                  "memory_type": memory_type,
              },
          )
      )

    logger.info(
        "Found %d memories for query '%s' (namespace=%s, user=%s)",
        len(memories),
        query[:50],
        namespace,
        user_id,
    )
    return SearchMemoryResponse(memories=memories)

  except Exception as e:
    logger.error("Failed to search memories: %s", e)
    return SearchMemoryResponse(memories=[])

close async

close() -> None

Close the memory service and cleanup resources.

Source code in src/adk_redis/memory/long_term_memory.py

async def close(self) -> None:
  """Close the memory service and cleanup resources."""
  if "_agent_memory_server_client" in self.__dict__:
    client = self.__dict__["_agent_memory_server_client"]
    close = getattr(client, "close", None)
    if close:
      await close()
    del self.__dict__["_agent_memory_server_client"]

RedisLongTermMemoryServiceConfig

Bases: BaseModel

Configuration for Redis long-term memory.

Attributes:

Name	Type	Description
backend	MemoryBackendName	Memory backend to use.
api_base_url	str	Memory API base URL.
api_key	str | None	Redis Agent Memory API key.
store_id	str | None	Redis Agent Memory store ID.
timeout	float	HTTP timeout in seconds.
timeout_ms	int | None	Optional SDK timeout in milliseconds.
default_namespace	str | None	Default namespace for memory operations.
search_top_k	int	Maximum number of memories to retrieve per search.
similarity_threshold	float | None	Minimum similarity threshold for search results.
distance_threshold	float | None	Backward-compatible alias for similarity_threshold.
store_events_as_messages	bool	Store ADK events as long-term message memories when add_session_to_memory or add_events_to_memory is called.
default_memory_type	MemoryTypeName	Default memory type for explicit add_memory writes.
default_topics	list[str]	Topics attached to created memory records.
source	str	Source label used when building deterministic IDs.
recency_boost	bool	Backward-compatible option retained from the previous Agent Memory Server client.
semantic_weight	float	Backward-compatible option retained from the previous Agent Memory Server client.
recency_weight	float	Backward-compatible option retained from the previous Agent Memory Server client.
freshness_weight	float	Backward-compatible option retained from the previous Agent Memory Server client.
novelty_weight	float	Backward-compatible option retained from the previous Agent Memory Server client.
half_life_last_access_days	float	Backward-compatible option retained from the previous Agent Memory Server client.
half_life_created_days	float	Backward-compatible option retained from the previous Agent Memory Server client.
extraction_strategy	Literal['discrete', 'summary', 'preferences', 'custom']	Backward-compatible option retained from the previous Agent Memory Server client.
extraction_strategy_config	dict[str, Any]	Backward-compatible option retained from the previous Agent Memory Server client.
model_name	str | None	Backward-compatible option retained from the previous Agent Memory Server client.
context_window_max	int | None	Backward-compatible option retained from the previous Agent Memory Server client.