Coverage for haystack/components/embedders/mock_text_embedder.py: 100%
50 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from __future__ import annotations
7from typing import Any
9from haystack import component, default_from_dict, default_to_dict
10from haystack.components.embedders.mock_utils import (
11 EmbeddingFn,
12 _coerce_embedding,
13 _deterministic_embedding,
14 _estimate_usage,
15)
16from haystack.utils import deserialize_callable, serialize_callable
19@component
20class MockTextEmbedder:
21 """
22 A Text Embedder that returns deterministic embeddings without calling any API.
24 It is a drop-in replacement for real Text Embedders (such as `OpenAITextEmbedder`) in tests, smoke tests, and
25 quick prototypes. It implements the same interface (`run`, `run_async`, serialization) but never contacts an
26 external service, so it is fully deterministic and free to run.
28 The embedding is selected based on how the component is configured:
30 - **Deterministic (default)**: with no configuration, the embedding is derived from a hash of the input text.
31 The same text always yields the same embedding, and different texts yield different embeddings, so the mock
32 works in retrieval pipelines and is reproducible across runs and processes.
33 - **Fixed embedding**: pass an `embedding` vector. The same vector is returned for every input.
34 - **Dynamic embedding**: pass an `embedding_fn` callable that receives the (prepared) text and returns the
35 embedding. This is useful when the embedding should depend on the input in a custom way.
37 ### Usage example
39 ```python
40 from haystack.components.embedders import MockTextEmbedder
42 embedder = MockTextEmbedder(dimension=8)
43 result = embedder.run("I love pizza!")
44 print(result["embedding"]) # a deterministic list of 8 floats
45 ```
46 """
48 def __init__(
49 self,
50 embedding: list[float] | None = None,
51 *,
52 embedding_fn: EmbeddingFn | None = None,
53 dimension: int = 768,
54 model: str = "mock-model",
55 meta: dict[str, Any] | None = None,
56 prefix: str = "",
57 suffix: str = "",
58 ) -> None:
59 """
60 Creates an instance of MockTextEmbedder.
62 :param embedding: An optional fixed embedding returned for every input. Mutually exclusive with
63 `embedding_fn`. If neither is provided, a deterministic embedding is derived from the input text.
64 :param embedding_fn: An optional callable that receives the prepared text (after `prefix`/`suffix` are
65 applied) and returns the embedding as a list of floats. Mutually exclusive with `embedding`. To support
66 serialization, pass a named function (lambdas and nested functions cannot be serialized).
67 :param dimension: The number of dimensions of the deterministic embedding. Ignored when `embedding` or
68 `embedding_fn` is provided, since their length is determined by the value or callable.
69 :param model: The model name reported in the metadata. Purely cosmetic; no model is loaded.
70 :param meta: Additional metadata merged into the output `meta`.
71 :param prefix: A string to add at the beginning of the text before embedding.
72 :param suffix: A string to add at the end of the text before embedding.
73 :raises ValueError: If both `embedding` and `embedding_fn` are provided, if `dimension` is not positive, or
74 if `embedding` is an empty list.
75 :raises TypeError: If `embedding` is not a sequence of numbers.
76 """
77 if embedding is not None and embedding_fn is not None:
78 raise ValueError("Pass either 'embedding' or 'embedding_fn', not both.")
79 if dimension <= 0:
80 raise ValueError("'dimension' must be a positive integer.")
82 self.embedding = _coerce_embedding(embedding, name="'embedding'") if embedding is not None else None
83 self.embedding_fn = embedding_fn
84 self.dimension = dimension
85 self.model = model
86 self.meta = meta or {}
87 self.prefix = prefix
88 self.suffix = suffix
89 self._is_warmed_up = False
91 def to_dict(self) -> dict[str, Any]:
92 """Serialize the component to a dictionary."""
93 embedding_fn = serialize_callable(self.embedding_fn) if self.embedding_fn is not None else None
94 return default_to_dict(
95 self,
96 embedding=self.embedding,
97 embedding_fn=embedding_fn,
98 dimension=self.dimension,
99 model=self.model,
100 meta=self.meta,
101 prefix=self.prefix,
102 suffix=self.suffix,
103 )
105 @classmethod
106 def from_dict(cls, data: dict[str, Any]) -> MockTextEmbedder:
107 """Deserialize the component from a dictionary."""
108 init_params = data.get("init_parameters", {})
109 embedding_fn = init_params.get("embedding_fn")
110 if embedding_fn:
111 init_params["embedding_fn"] = deserialize_callable(embedding_fn)
112 return default_from_dict(cls, data)
114 def warm_up(self) -> None:
115 """No-op warm up, provided for interface compatibility with real Embedders."""
116 self._is_warmed_up = True
118 def _embed(self, text: str) -> list[float]:
119 """Produce the embedding for the prepared text according to the configured mode."""
120 if self.embedding_fn is not None:
121 return _coerce_embedding(self.embedding_fn(text), name="the return value of 'embedding_fn'")
122 if self.embedding is not None:
123 return list(self.embedding)
124 return _deterministic_embedding(text, self.dimension)
126 @component.output_types(embedding=list[float], meta=dict[str, Any])
127 def run(self, text: str) -> dict[str, Any]:
128 """
129 Return a deterministic embedding for the input text without calling any API.
131 :param text: The text to embed.
132 :returns: A dictionary with the following keys:
133 - `embedding`: The embedding of the input text.
134 - `meta`: Metadata about the (mock) model.
135 :raises TypeError: If `text` is not a string.
136 """
137 self.warm_up()
139 if not isinstance(text, str):
140 raise TypeError(
141 "MockTextEmbedder expects a string as an input. "
142 "In case you want to embed a list of Documents, please use the MockDocumentEmbedder."
143 )
145 text_to_embed = self.prefix + text + self.suffix
146 meta: dict[str, Any] = {"model": self.model, "usage": _estimate_usage([text_to_embed])}
147 meta.update(self.meta)
148 return {"embedding": self._embed(text_to_embed), "meta": meta}
150 @component.output_types(embedding=list[float], meta=dict[str, Any])
151 async def run_async(self, text: str) -> dict[str, Any]:
152 """
153 Asynchronously return a deterministic embedding for the input text without calling any API.
155 :param text: The text to embed.
156 :returns: A dictionary with the following keys:
157 - `embedding`: The embedding of the input text.
158 - `meta`: Metadata about the (mock) model.
159 :raises TypeError: If `text` is not a string.
160 """
161 return self.run(text=text)