Class InMemoryMemoryService

src/google/adk/memory/in_memory_memory_service.py:45–134

An in-memory memory service for prototyping purpose only. Uses keyword matching instead of semantic search. This class is thread-safe, however, it should be used for testing and development only.

class InMemoryMemoryService(BaseMemoryService):
  """An in-memory memory service for prototyping purpose only.

  Uses keyword matching instead of semantic search.

  This class is thread-safe, however, it should be used for testing and
  development only.
  """

  def __init__(self):
    self._lock = threading.Lock()

    self._session_events: dict[str, dict[str, list[Event]]] = {}
    """Keys are "{app_name}/{user_id}". Values are dicts of session_id to
    session event lists.
    """

  @override
  async def add_session_to_memory(self, session: Session) -> None:
    user_key = _user_key(session.app_name, session.user_id)

    with self._lock:
      self._session_events[user_key] = self._session_events.get(user_key, {})
      self._session_events[user_key][session.id] = [
          event
          for event in session.events
          if event.content and event.content.parts
      ]

  @override
  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:
    _ = custom_metadata
    user_key = _user_key(app_name, user_id)
    scoped_session_id = session_id or _UNKNOWN_SESSION_ID
    events_to_add = [
        event for event in events if event.content and event.content.parts
    ]

    with self._lock:
      self._session_events[user_key] = self._session_events.get(user_key, {})
      existing_events = self._session_events[user_key].get(
          scoped_session_id, []
      )
      existing_ids = {event.id for event in existing_events}
      for event in events_to_add:
        if event.id not in existing_ids:
          existing_events.append(event)
          existing_ids.add(event.id)
      self._session_events[user_key][scoped_session_id] = existing_events