async def search_memory(
      self, *, app_name: str, user_id: str, query: str
  ) -> SearchMemoryResponse:
    """Searches for sessions that match the query using rag.retrieval_query."""
    from ..events.event import Event

    response = rag.retrieval_query(
        text=query,
        rag_resources=self._vertex_rag_store.rag_resources,
        rag_corpora=self._vertex_rag_store.rag_corpora,
        similarity_top_k=self._vertex_rag_store.similarity_top_k,
        vector_distance_threshold=self._vertex_rag_store.vector_distance_threshold,
    )

    memory_results = []
    session_events_map = OrderedDict()
    for context in response.contexts.contexts:
      # filter out context that is not related
      # TODO: Add server side filtering by app_name and user_id.
      if not context.source_display_name.startswith(f"{app_name}.{user_id}."):
        continue
      session_id = context.source_display_name.split(".")[-1]
      events = []
      if context.text:
        lines = context.text.split("\n")

        for line in lines:
          line = line.strip()
          if not line:
            continue

          try:
            # Try to parse as JSON
            event_data = json.loads(line)

            author = event_data.get("author", "")
            timestamp = float(event_data.get("timestamp", 0))
            text = event_data.get("text", "")

            content = types.Content(parts=[types.Part(text=text)])
            event = Event(author=author, timestamp=timestamp, content=content)
            events.append(event)
          except json.JSONDecodeError:
            # Not valid JSON, skip this line
            continue

      if session_id in session_events_map:
        session_events_map[session_id].append(events)
      else:
        session_events_map[session_id] = [events]

    # Remove overlap and combine events from the same session.
    for session_id, event_lists in session_events_map.items():
      for events in _merge_event_lists(event_lists):
        sorted_events = sorted(events, key=lambda e: e.timestamp)

        memory_results.extend([
            MemoryEntry(
                author=event.author,
                content=event.content,
                timestamp=_utils.format_timestamp(event.timestamp),
            )
            for event in sorted_events
            if event.content
        ])
    return SearchMemoryResponse(memories=memory_results)
Searches for sessions that match the query using rag.retrieval_query.
src/google/adk/memory/vertex_ai_rag_memory_service.py