Repository: deepset-ai/haystack

Issue:
Weights and score normalization for `JoinDocuments` node with reciprocal rank fusion (PR ready)
**Is your feature request related to a problem? Please describe.**
This PR would address two existing problems with using the `JoinDocuments` node with reciprocal rank fusion (RRF):
1. The node's `weights` are not applied with RRF, and input rankings are always weighted equally.
2. Scores are not normalized. As a result, the output documents receive their raw RRF scores, which are not very useful.

**Describe the solution you'd like**
I've got a PR ready that weights and normalizes the RRF scores. As it's currently implemented, the normalization would always happen--I've yet to see a use case where we'd want the raw RRF scores--but the code can easily be revised so that normalization is optional. The changes would be non-breaking in the sense that if no weights are provided (= all input nodes are weighted equally), the output documents would be identical to those currently output by the `JoinDocuments` node (_modulo_ score normalization).

**Describe alternatives you've considered**
Not really too many alternatives to this approach beyond custom nodes, as far as I can tell. There would be some small decisions with this approach, namely, is normalization applied automatically or by choice?

**Additional context**
Hybrid search pipelines often make use of `JoinDocuments` nodes with RRF, so ensuring that weights can be applied and scores are interpretable would be quite useful.


Discussion:
coveralls: ## Pull Request Test Coverage Report for [Build 6225504685](https://coveralls.io/builds/62728250)

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.01%**) to **49.062%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/62728250/badge)](https://coveralls.io/builds/62728250) |
| :-- | --: |
| Change from base [Build 6219800940](https://coveralls.io/builds/62714306): |  0.01% |
| Covered Lines: | 11847 |
| Relevant Lines: | 24147 |

---
##### 💛  - [Coveralls](https://coveralls.io)

anakin87: Hey, @robpasternak...

Thanks for this contribution! :blue_heart: 

1) Please add a release note about this new feature as described [here](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#release-notes).
2) Can you add 1 or 2 unit tests [here](https://github.com/deepset-ai/haystack/blob/main/test/nodes/test_join_documents.py), to make sure that this new feature performs as expected?
robpasternak: Will do, and I'll let you know when it's done. Thanks @anakin87!
Timoeller: Please also use standard syntax when linking issues @robpasternak . Because when you use "adresses" it does not automatically link PR and issue, if you use "fixes" it does 🎉 
anakin87: Hey, @robpasternak...

Do you think you can add the release note and a unit test or two?

if you can't, we'll do it when possible...
robpasternak: @anakin87 Sorry, this slipped off the docket. I'll get that done tomorrow.
anakin87: Thanks, @robpasternak!

Please make sure to merge the v1.x branch.
I remember that there has been work on this topic: see #6261.
anakin87: @robpasternak I took the liberty of rebasing your branch.
Please go on from there...

Patch:
diff --git a/haystack/nodes/other/join_docs.py b/haystack/nodes/other/join_docs.py
index 274e90a3..b94d7d14 100644
--- a/haystack/nodes/other/join_docs.py
+++ b/haystack/nodes/other/join_docs.py
@@ -155,8 +155,16 @@ class JoinDocuments(JoinNode):
         K = 61
 
         scores_map = defaultdict(int)
-        for result in results:
+        weights = self.weights if self.weights else [1 / len(results)] * len(results)
+
+        # Calculate weighted reciprocal rank fusion score
+        for result, weight in zip(results, weights):
             for rank, doc in enumerate(result):
-                scores_map[doc.id] += 1 / (K + rank)
+                scores_map[doc.id] += (weight * len(results)) / (K + rank)
+
+        # Normalize scores. Note: len(results) / K is the maximum possible score,
+        # achieved by being ranked first in all results with non-zero weight.
+        for id in scores_map:
+            scores_map[id] = scores_map[id] / (len(results) / K)
 
         return scores_map
diff --git a/haystack/nodes/retriever/_embedding_encoder.py b/haystack/nodes/retriever/_embedding_encoder.py
index d07e7ab1..f75b9971 100644
--- a/haystack/nodes/retriever/_embedding_encoder.py
+++ b/haystack/nodes/retriever/_embedding_encoder.py
@@ -16,7 +16,7 @@ from haystack.environment import (
     HAYSTACK_REMOTE_API_MAX_RETRIES,
     HAYSTACK_REMOTE_API_TIMEOUT_SEC,
 )
-from haystack.errors import CohereError, CohereUnauthorizedError
+from haystack.errors import AWSConfigurationError, CohereError, CohereUnauthorizedError
 from haystack.nodes.retriever._openai_encoder import _OpenAIEmbeddingEncoder
 from haystack.schema import Document
 from haystack.telemetry import send_event
@@ -42,6 +42,9 @@ with LazyImport(message="Run 'pip install farm-haystack[inference]'") as torch_a
     from haystack.modeling.infer import Inferencer
     from haystack.nodes.retriever._losses import _TRAINING_LOSSES
 
+with LazyImport(message="Run 'pip install boto3'") as boto3_import:
+    import boto3
+    from botocore.exceptions import BotoCoreError
 
 COHERE_TIMEOUT = float(os.environ.get(HAYSTACK_REMOTE_API_TIMEOUT_SEC, 30))
 COHERE_BACKOFF = int(os.environ.get(HAYSTACK_REMOTE_API_BACKOFF_SEC, 10))
@@ -55,6 +58,8 @@ COHERE_EMBEDDING_MODELS = [
     "embed-multilingual-v2.0",
 ]
 
+BEDROCK_EMBEDDING_MODELS = ["amazon.titan-embed-text-v1", "cohere.embed-english-v3", "cohere.embed-multilingual-v3"]
+
 
 class _DefaultEmbeddingEncoder(_BaseEmbeddingEncoder):
     def __init__(self, retriever: "EmbeddingRetriever"):
@@ -434,6 +439,107 @@ class _CohereEmbeddingEncoder(_BaseEmbeddingEncoder):
         raise NotImplementedError(f"Saving is not implemented for {self.__class__}")
 
 
+class _BedrockEmbeddingEncoder(_BaseEmbeddingEncoder):
+    def __init__(self, retriever: "EmbeddingRetriever"):
+        """Embedding Encoder for Bedrock models
+        See https://docs.aws.amazon.com/bedrock/latest/userguide/embeddings.html for more details.
+        The maximum input text is 8K tokens and the maximum output vector length is 1536.
+        Titan embeddings do not support batch operations.
+
+        :param retriever: EmbeddingRetriever object
+        """
+        boto3_import.check()
+        if retriever.embedding_model not in BEDROCK_EMBEDDING_MODELS:
+            raise ValueError("Model not supported by Bedrock Embedding Encoder")
+        self.model = retriever.embedding_model
+        self.client = self._initialize_boto3_session(retriever.aws_config).client("bedrock-runtime")
+
+    def _initialize_boto3_session(self, aws_config: Optional[Dict[str, Any]]):
+        if aws_config is None:
+            raise ValueError(
+                "`aws_config` is not set. To use Bedrock models, you should set `aws_config` when initializing the retriever."
+            )
+
+        aws_access_key_id = aws_config.get("aws_access_key_id", None)
+        aws_secret_access_key = aws_config.get("aws_secret_access_key", None)
+        aws_session_token = aws_config.get("aws_session_token", None)
+        region_name = aws_config.get("region_name", None)
+        profile_name = aws_config.get("profile_name", None)
+        try:
+            return boto3.Session(
+                aws_access_key_id=aws_access_key_id,
+                aws_secret_access_key=aws_secret_access_key,
+                aws_session_token=aws_session_token,
+                region_name=region_name,
+                profile_name=profile_name,
+            )
+        except BotoCoreError as e:
+            raise AWSConfigurationError(
+                f"Failed to initialize the session with provided AWS credentials {aws_config}"
+            ) from e
+
+    def _embed_batch_cohere(
+        self, texts: List[str], input_type: Literal["search_query", "search_document"]
+    ) -> np.ndarray:
+        cohere_payload = {"texts": texts, "input_type": input_type, "truncate": "RIGHT"}
+        response = self._invoke_model(cohere_payload)
+        embeddings = np.array(response["embeddings"])
+        return embeddings
+
+    def _embed_titan(self, text: str) -> np.ndarray:
+        titan_payload = {"inputText": text}
+        response = self._invoke_model(titan_payload)
+        embeddings = np.array(response["embedding"])
+        return embeddings
+
+    def _invoke_model(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        body = json.dumps(payload)
+        response = self.client.invoke_model(
+            body=body, modelId=self.model, accept="application/json", contentType="application/json"
+        )
+        body = response.get("body").read().decode("utf-8")
+        response_body = json.loads(body)
+        return response_body
+
+    def embed_queries(self, queries: List[str]) -> np.ndarray:
+        if self.model == "amazon.titan-embed-text-v1":
+            all_embeddings = []
+            for query in queries:
+                generated_embeddings = self._embed_titan(query)
+                all_embeddings.append(generated_embeddings)
+            return np.stack(all_embeddings)
+        else:
+            return self._embed_batch_cohere(queries, input_type="search_query")
+
+    def embed_documents(self, docs: List[Document]) -> np.ndarray:
+        if self.model == "amazon.titan-embed-text-v1":
+            all_embeddings = []
+            for doc in docs:
+                generated_embeddings = self._embed_titan(doc.content)
+                all_embeddings.append(generated_embeddings)
+            return np.stack(all_embeddings)
+        else:
+            contents = [d.content for d in docs]
+            return self._embed_batch_cohere(contents, input_type="search_document")
+
+    def train(
+        self,
+        training_data: List[Dict[str, Any]],
+        learning_rate: float = 2e-5,
+        n_epochs: int = 1,
+        num_warmup_steps: Optional[int] = None,
+        batch_size: int = 16,
+        train_loss: Literal["mnrl", "margin_mse"] = "mnrl",
+        num_workers: int = 0,
+        use_amp: bool = False,
+        **kwargs,
+    ):
+        raise NotImplementedError(f"Training is not implemented for {self.__class__}")
+
+    def save(self, save_dir: Union[Path, str]):
+        raise NotImplementedError(f"Saving is not implemented for {self.__class__}")
+
+
 _EMBEDDING_ENCODERS: Dict[str, Callable] = {
     "farm": _DefaultEmbeddingEncoder,
     "transformers": _DefaultEmbeddingEncoder,
@@ -441,4 +547,5 @@ _EMBEDDING_ENCODERS: Dict[str, Callable] = {
     "retribert": _RetribertEmbeddingEncoder,
     "openai": _OpenAIEmbeddingEncoder,
     "cohere": _CohereEmbeddingEncoder,
+    "bedrock": _BedrockEmbeddingEncoder,
 }
diff --git a/haystack/nodes/retriever/dense.py b/haystack/nodes/retriever/dense.py
index d4663e6c..3139ec15 100644
--- a/haystack/nodes/retriever/dense.py
+++ b/haystack/nodes/retriever/dense.py
@@ -17,7 +17,11 @@ from haystack.errors import HaystackError
 from haystack.schema import Document, FilterType
 from haystack.document_stores import BaseDocumentStore
 from haystack.nodes.retriever.base import BaseRetriever
-from haystack.nodes.retriever._embedding_encoder import _EMBEDDING_ENCODERS, COHERE_EMBEDDING_MODELS
+from haystack.nodes.retriever._embedding_encoder import (
+    _EMBEDDING_ENCODERS,
+    COHERE_EMBEDDING_MODELS,
+    BEDROCK_EMBEDDING_MODELS,
+)
 from haystack.utils.early_stopping import EarlyStopping
 from haystack.telemetry import send_event
 from haystack.lazy_imports import LazyImport
@@ -1468,13 +1472,15 @@ class EmbeddingRetriever(DenseRetriever):
         azure_deployment_name: Optional[str] = None,
         api_base: str = "https://api.openai.com/v1",
         openai_organization: Optional[str] = None,
+        aws_config: Optional[Dict[str, Any]] = None,
     ):
         """
         :param document_store: An instance of DocumentStore from which to retrieve documents.
         :param embedding_model: Local path or name of model in Hugging Face's model hub such
                                 as ``'sentence-transformers/all-MiniLM-L6-v2'``. The embedding model could also
                                 potentially be an OpenAI model ["ada", "babbage", "davinci", "curie"] or
-                                a Cohere model ["embed-english-v2.0", "embed-english-light-v2.0", "embed-multilingual-v2.0"].
+                                a Cohere model ["embed-english-v2.0", "embed-english-light-v2.0", "embed-multilingual-v2.0"] or
+                                an AWS Bedrock model ["amazon.titan-embed-text-v1", "cohere.embed-english-v3", "cohere.embed-multilingual-v3"].
         :param model_version: The version of model to use from the HuggingFace model hub. Can be tag name, branch name, or commit hash.
         :param use_gpu: Whether to use all available GPUs or the CPU. Falls back on CPU if no GPU is available.
         :param batch_size: Number of documents to encode at once.
@@ -1489,6 +1495,7 @@ class EmbeddingRetriever(DenseRetriever):
         4. `retribert` : (will use `_RetribertEmbeddingEncoder` as embedding encoder)
         5. `openai` : (will use `_OpenAIEmbeddingEncoder` as embedding encoder)
         6. `cohere` : (will use `_CohereEmbeddingEncoder` as embedding encoder)
+        7. `bedrock` : (will use `_BedrockEmbeddingEncoder` as embedding encoder)
 
         :param pooling_strategy: Strategy for combining the embeddings from the model (for farm / transformers models only).
                                  Options:
@@ -1533,6 +1540,7 @@ class EmbeddingRetriever(DenseRetriever):
         :param api_base: The OpenAI API base URL, defaults to `"https://api.openai.com/v1"`.
         :param openai_organization: The OpenAI-Organization ID, defaults to `None`. For more details, see OpenAI
         [documentation](https://platform.openai.com/docs/api-reference/requesting-organization).
+        :param aws_config: The aws_config contains {aws_access_key, aws_secret_key, aws_region, profile_name} to use with the boto3 Session for an AWS Bedrock retriever. Defaults to 'None'.
         """
         torch_and_transformers_import.check()
 
@@ -1565,6 +1573,7 @@ class EmbeddingRetriever(DenseRetriever):
         self.azure_base_url = azure_base_url
         self.azure_deployment_name = azure_deployment_name
         self.openai_organization = openai_organization
+        self.aws_config = aws_config
         self.model_format = (
             self._infer_model_format(model_name_or_path=embedding_model, use_auth_token=use_auth_token)
             if model_format is None
@@ -1885,6 +1894,7 @@ class EmbeddingRetriever(DenseRetriever):
 
     @staticmethod
     def _infer_model_format(model_name_or_path: str, use_auth_token: Optional[Union[str, bool]]) -> str:
+        # pylint: disable=too-many-return-statements
         valid_openai_model_name = model_name_or_path in ["ada", "babbage", "davinci", "curie"] or any(
             m in model_name_or_path for m in ["-ada-", "-babbage-", "-davinci-", "-curie-"]
         )
@@ -1892,6 +1902,8 @@ class EmbeddingRetriever(DenseRetriever):
             return "openai"
         if model_name_or_path in COHERE_EMBEDDING_MODELS:
             return "cohere"
+        if model_name_or_path in BEDROCK_EMBEDDING_MODELS:
+            return "bedrock"
         # Check if model name is a local directory with sentence transformers config file in it
         if Path(model_name_or_path).exists():
             if Path(f"{model_name_or_path}/config_sentence_transformers.json").exists():
diff --git a/releasenotes/notes/aws-bedrock-embedding-encoder-a978884c1a2c8237.yaml b/releasenotes/notes/aws-bedrock-embedding-encoder-a978884c1a2c8237.yaml
new file mode 100644
index 00000000..caa1f479
--- /dev/null
+++ b/releasenotes/notes/aws-bedrock-embedding-encoder-a978884c1a2c8237.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Adding Bedrock Embeddings Encoder to use as a retriever.
diff --git a/releasenotes/notes/join-docs-weighting-rrf-c52ba00a25004fd4.yaml b/releasenotes/notes/join-docs-weighting-rrf-c52ba00a25004fd4.yaml
new file mode 100644
index 00000000..23a31d91
--- /dev/null
+++ b/releasenotes/notes/join-docs-weighting-rrf-c52ba00a25004fd4.yaml
@@ -0,0 +1,6 @@
+---
+enhancements:
+  - |
+    Make `JoinDocuments` sensitive to `weights` parameter when
+    `join_mode` is reciprocal rank fusion. Add score normalization
+    for `JoinDocuments` when `join_mode` is reciprocal rank fusion.


Test patch:
diff --git a/test/nodes/test_join_documents.py b/test/nodes/test_join_documents.py
index 463aeaa5..ae809f49 100644
--- a/test/nodes/test_join_documents.py
+++ b/test/nodes/test_join_documents.py
@@ -3,6 +3,7 @@ import pytest
 
 from haystack import Document
 from haystack.nodes.other.join_docs import JoinDocuments
+from copy import deepcopy
 
 
 @pytest.mark.unit
@@ -113,3 +114,38 @@ def test_joindocuments_concatenate_duplicate_docs_null_score():
     result, _ = join_docs.run(inputs)
     assert len(result["documents"]) == 3
     assert result["documents"] == expected_outputs["documents"]
+
+
+@pytest.mark.unit
+def test_joindocuments_rrf_weights():
+    """
+    Test that the reciprocal rank fusion method correctly handles weights.
+    """
+    inputs_none = [
+        {
+            "documents": [
+                Document(content="text document 1", content_type="text", score=0.2),
+                Document(content="text document 2", content_type="text", score=0.3),
+            ]
+        },
+        {
+            "documents": [
+                Document(content="text document 3", content_type="text", score=0.7),
+                Document(content="text document 4", content_type="text", score=None),
+            ]
+        },
+    ]
+
+    inputs_even = deepcopy(inputs_none)
+    inputs_uneven = deepcopy(inputs_none)
+
+    join_docs_none = JoinDocuments(join_mode="reciprocal_rank_fusion")
+    result_none, _ = join_docs_none.run(inputs_none)
+    join_docs_even = JoinDocuments(join_mode="reciprocal_rank_fusion", weights=[0.5, 0.5])
+    result_even, _ = join_docs_even.run(inputs_even)
+    join_docs_uneven = JoinDocuments(join_mode="reciprocal_rank_fusion", weights=[0.7, 0.3])
+    result_uneven, _ = join_docs_uneven.run(inputs_uneven)
+
+    assert result_none["documents"] == result_even["documents"]
+    assert result_uneven["documents"] != result_none["documents"]
+    assert result_uneven["documents"][0].score > result_none["documents"][0].score