Coverage for haystack/components/embedders/mock_utils.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5import hashlib 

6import math 

7import random 

8from collections.abc import Callable 

9 

10# A callable that derives an embedding from the (prepared) text to embed. It receives the text and returns the 

11# embedding as a list of floats. 

12EmbeddingFn = Callable[[str], list[float]] 

13 

14 

15def _l2_normalize(vector: list[float]) -> list[float]: 

16 """Return the L2-normalized vector, so that mock embeddings behave like real (unit-length) ones.""" 

17 norm = math.sqrt(sum(value * value for value in vector)) 

18 if norm == 0.0: 

19 return vector 

20 return [value / norm for value in vector] 

21 

22 

23def _deterministic_embedding(text: str, dimension: int) -> list[float]: 

24 """ 

25 Generate a deterministic, unit-length embedding from the given text. 

26 

27 The same text always yields the same embedding, and different texts yield different embeddings, which makes mock 

28 embeddings usable in retrieval pipelines and reproducible across runs and processes. The seed is derived from a 

29 SHA-256 digest of the text (not the process-salted built-in `hash`) to guarantee cross-process stability. 

30 

31 :param text: The text to embed. 

32 :param dimension: The number of dimensions of the resulting embedding. 

33 :returns: A deterministic, L2-normalized embedding of length `dimension`. 

34 """ 

35 digest = hashlib.sha256(text.encode("utf-8")).digest() 

36 seed = int.from_bytes(digest[:8], "big") 

37 rng = random.Random(seed) 

38 vector = [rng.uniform(-1.0, 1.0) for _ in range(dimension)] 

39 return _l2_normalize(vector) 

40 

41 

42def _coerce_embedding(value: object, *, name: str) -> list[float]: 

43 """ 

44 Validate that `value` is a non-empty sequence of numbers and coerce it into a list of floats. 

45 

46 :param value: The value to validate, e.g. a user-provided fixed embedding or the output of an `embedding_fn`. 

47 :param name: How to refer to `value` in error messages, e.g. ``"'embedding'"``. 

48 """ 

49 if not isinstance(value, (list, tuple)) or not all(isinstance(item, (int, float)) for item in value): 

50 raise TypeError(f"{name} must be a sequence of numbers, got {type(value)}.") 

51 if len(value) == 0: 

52 raise ValueError(f"{name} must not be empty.") 

53 return [float(item) for item in value] 

54 

55 

56def _estimate_usage(texts: list[str]) -> dict[str, int]: 

57 """ 

58 Roughly estimate token usage as whitespace-separated word counts. 

59 

60 This is an approximation (not real tokenization) intended to give downstream code realistic-looking metadata. 

61 """ 

62 prompt_tokens = sum(len(text.split()) for text in texts) 

63 return {"prompt_tokens": prompt_tokens, "total_tokens": prompt_tokens}